Repository: Tochemey/goakt
Pull request #213 (2024-01-15 21:11:16)

feat: add cluster events
https://github.com/Tochemey/goakt/issues/211

- [x] Add the implementation of emitting the events.
- [x] Hook into the actor system's events stream
- [x] Add unit tests on Cluster code
- [x] Add unit test on the actor system code

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/213?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
Attention: `35 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`ca9b73a`)](https://app.codecov.io/gh/Tochemey/goakt/commit/ca9b73afb030883ad81edf0ca11baee6708ac09a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 88.71% compared to head [(`c7327a3`)](https://app.codecov.io/gh/Tochemey/goakt/pull/213?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 88.12%.

| [Files](https://app.codecov.io/gh/Tochemey/goakt/pull/213?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) | Patch % | Lines |
|---|---|---|
| [cluster/cluster.go](https://app.codecov.io/gh/Tochemey/goakt/pull/213?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-Y2x1c3Rlci9jbHVzdGVyLmdv) | 32.69% | [32 Missing and 3 partials :warning: ](https://app.codecov.io/gh/Tochemey/goakt/pull/213?src=pr&el=tree&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     #213      +/-   ##
==========================================
- Coverage   88.71%   88.12%   -0.60%     
==========================================
  Files          44       44              
  Lines        4920     4970      +50     
==========================================
+ Hits         4365     4380      +15     
- Misses        439      471      +32     
- Partials      116      119       +3     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/213?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/actor_system.go b/actors/actor_system.go
index 9ce6a46..da0078c 100644
--- a/actors/actor_system.go
+++ b/actors/actor_system.go
@@ -109,6 +109,9 @@ type ActorSystem interface {
 	ScheduleWithCron(ctx context.Context, message proto.Message, pid PID, cronExpression string) error
 	// RemoteScheduleWithCron schedules a message to be sent to an actor in the future using a cron expression.
 	RemoteScheduleWithCron(ctx context.Context, message proto.Message, address *addresspb.Address, cronExpression string) error
+	// PeerAddress returns the actor system address known in the cluster. That address is used by other nodes to communicate with the actor system.
+	// This address is empty when cluster mode is not activated
+	PeerAddress() string
 	// handleRemoteAsk handles a synchronous message to another actor and expect a response.
 	// This block until a response is received or timed out.
 	handleRemoteAsk(ctx context.Context, to PID, message proto.Message) (response proto.Message, err error)
@@ -191,6 +194,7 @@ type actorSystem struct {
 	tracer       trace.Tracer
 	// specifies whether metric is enabled
 	metricEnabled atomic.Bool
+	eventsChan    <-chan *cluster.Event
 }
 
 // enforce compilation error when all methods of the ActorSystem interface are not implemented
@@ -227,6 +231,7 @@ func NewActorSystem(name string, opts ...Option) (ActorSystem, error) {
 		partitionHasher:     hash.DefaultHasher(),
 		actorInitTimeout:    DefaultInitTimeout,
 		tracer:              noop.NewTracerProvider().Tracer(name),
+		eventsChan:          make(chan *cluster.Event, 1),
 	}
 	// set the atomic settings
 	system.hasStarted.Store(false)
@@ -548,6 +553,21 @@ func (x *actorSystem) Actors() []PID {
 	return x.actors.List()
 }
 
+// PeerAddress returns the actor system address known in the cluster. That address is used by other nodes to communicate with the actor system.
+// This address is empty when cluster mode is not activated
+func (x *actorSystem) PeerAddress() string {
+	// acquire the lock
+	x.sem.Lock()
+	// release the lock
+	defer x.sem.Unlock()
+	// check whether cluster mode is enabled
+	if x.clusterEnabled.Load() {
+		// return the cluster node advertised address
+		return x.cluster.AdvertisedAddress()
+	}
+	return ""
+}
+
 // ActorOf returns an existing actor in the local system or in the cluster when clustering is enabled
 // When cluster mode is activated, the PID will be nil.
 // When remoting is enabled this method will return and error
@@ -810,10 +830,8 @@ func (x *actorSystem) Stop(ctx context.Context) error {
 		}
 		// stop broadcasting cluster messages
 		close(x.clusterChan)
-
 		// unset the remoting settings
 		x.clusterEnabled.Store(false)
-		x.cluster = nil
 	}
 
 	// short-circuit the shutdown process when there are no online actors
@@ -1106,7 +1124,7 @@ func (x *actorSystem) enableClustering(ctx context.Context) {
 	x.logger.Info("enabling clustering...")
 
 	// create an instance of the cluster service and start it
-	cluster, err := cluster.New(x.Name(),
+	cluster, err := cluster.NewNode(x.Name(),
 		x.serviceDiscovery,
 		cluster.WithLogger(x.logger),
 		cluster.WithPartitionsCount(x.partitionsCount),
@@ -1139,11 +1157,15 @@ func (x *actorSystem) enableClustering(ctx context.Context) {
 	x.sem.Lock()
 	// set the cluster field
 	x.cluster = cluster
+	// set the cluster events channel
+	x.eventsChan = cluster.Events()
 	// set the remoting host and port
 	x.remotingHost = cluster.NodeHost()
 	x.remotingPort = int32(cluster.NodeRemotingPort())
 	// release the lock
 	x.sem.Unlock()
+	// start listening to cluster events
+	go x.broadcastClusterEvents()
 	// start broadcasting cluster message
 	go x.broadcast(ctx)
 	// add some logging
@@ -1241,6 +1263,7 @@ func (x *actorSystem) reset() {
 	x.telemetry = nil
 	x.actors = newPIDMap(10)
 	x.name = ""
+	x.cluster = nil
 }
 
 // broadcast publishes newly created actor into the cluster when cluster is enabled
@@ -1323,3 +1346,27 @@ func (x *actorSystem) registerMetrics() error {
 
 	return err
 }
+
+// broadcastClusterEvents listens to cluster events
+func (x *actorSystem) broadcastClusterEvents() {
+	// read from the channel
+	for event := range x.eventsChan {
+		// when cluster is enabled
+		if x.clusterEnabled.Load() {
+			// only push cluster event when defined
+			if event != nil && event.Type != nil {
+				// unpack the event
+				message, _ := event.Type.UnmarshalNew()
+				// push the cluster event when event stream is set
+				if x.eventsStream != nil {
+					// add some debug log
+					x.logger.Debugf("node=(%s) publishing cluster event=(%s)....", x.name, event.Type.GetTypeUrl())
+					// send the event to the event streams
+					x.eventsStream.Publish(eventsTopic, message)
+					// add some debug log
+					x.logger.Debugf("cluster event=(%s) successfully published by node=(%s)", event.Type.GetTypeUrl(), x.name)
+				}
+			}
+		}
+	}
+}
diff --git a/actors/scheduler.go b/actors/scheduler.go
index 2d4c23f..0454fb8 100644
--- a/actors/scheduler.go
+++ b/actors/scheduler.go
@@ -26,13 +26,16 @@ package actors
 
 import (
 	"context"
+	"fmt"
+	"net"
 	"strconv"
 	"sync"
 	"time"
 
 	"github.com/pkg/errors"
+	"github.com/reugn/go-quartz/job"
+	quartzlogger "github.com/reugn/go-quartz/logger"
 	"github.com/reugn/go-quartz/quartz"
-	quartzlogger "github.com/reugn/go-quartz/quartz/logger"
 	"github.com/tochemey/goakt/cluster"
 	"github.com/tochemey/goakt/log"
 	addresspb "github.com/tochemey/goakt/pb/address/v1"
@@ -116,7 +119,7 @@ func (x *scheduler) Stop(ctx context.Context) {
 	// release the lock once done
 	defer x.mu.Unlock()
 	// clear all scheduled jobs
-	x.quartzScheduler.Clear()
+	_ = x.quartzScheduler.Clear()
 	// stop the scheduler
 	x.quartzScheduler.Stop()
 	// set the started
@@ -144,7 +147,7 @@ func (x *scheduler) ScheduleOnce(ctx context.Context, message proto.Message, pid
 		return errors.New("messages scheduler is not started")
 	}
 	// create the job
-	job := quartz.NewFunctionJob[bool](func(ctx context.Context) (bool, error) {
+	job := job.NewFunctionJob[bool](func(ctx context.Context) (bool, error) {
 		// when the job run send the message to actor
 		if err := Tell(ctx, pid, message); err != nil {
 			return false, err
@@ -153,8 +156,11 @@ func (x *scheduler) ScheduleOnce(ctx context.Context, message proto.Message, pid
 		return true, nil
 	})
 
+	// create the job details
+	jobDetails := quartz.NewJobDetail(job, quartz.NewJobKey(pid.ActorPath().String()))
+
 	// check whether the job is already scheduled
-	if err := x.distributeJobKeyOrNot(ctx, job); err != nil {
+	if err := x.distributeJobKeyOrNot(ctx, jobDetails); err != nil {
 		// skip the job scheduling when the key is already distributed
 		if errors.Is(err, errSkipJobScheduling) {
 			return nil
@@ -163,7 +169,7 @@ func (x *scheduler) ScheduleOnce(ctx context.Context, message proto.Message, pid
 	}
 
 	// schedule the job
-	return x.quartzScheduler.ScheduleJob(ctx, job, quartz.NewRunOnceTrigger(interval))
+	return x.quartzScheduler.ScheduleJob(jobDetails, quartz.NewRunOnceTrigger(interval))
 }
 
 // RemoteScheduleOnce schedules a message to be sent to a remote actor in the future.
@@ -182,7 +188,7 @@ func (x *scheduler) RemoteScheduleOnce(ctx context.Context, message proto.Messag
 		return errors.New("messages scheduler is not started")
 	}
 	// create the job
-	job := quartz.NewFunctionJob[bool](func(ctx context.Context) (bool, error) {
+	job := job.NewFunctionJob[bool](func(ctx context.Context) (bool, error) {
 		// when the job run send the message to actor
 		if err := RemoteTell(ctx, address, message); err != nil {
 			return false, err
@@ -190,9 +196,12 @@ func (x *scheduler) RemoteScheduleOnce(ctx context.Context, message proto.Messag
 		// return true when successful
 		return true, nil
 	})
-
+	// create the job key name
+	key := fmt.Sprintf("%s@%s", address.GetName(), net.JoinHostPort(address.GetHost(), strconv.Itoa(int(address.GetPort()))))
+	// create the job details
+	jobDetails := quartz.NewJobDetail(job, quartz.NewJobKey(key))
 	// check whether the job is already scheduled
-	if err := x.distributeJobKeyOrNot(ctx, job); err != nil {
+	if err := x.distributeJobKeyOrNot(ctx, jobDetails); err != nil {
 		// skip the job scheduling when the key is already distributed
 		if errors.Is(err, errSkipJobScheduling) {
 			return nil
@@ -200,7 +209,7 @@ func (x *scheduler) RemoteScheduleOnce(ctx context.Context, message proto.Messag
 		return err
 	}
 	// schedule the job
-	return x.quartzScheduler.ScheduleJob(ctx, job, quartz.NewRunOnceTrigger(interval))
+	return x.quartzScheduler.ScheduleJob(jobDetails, quartz.NewRunOnceTrigger(interval))
 }
 
 // ScheduleWithCron schedules a message to be sent to an actor in the future using a cron expression.
@@ -214,7 +223,7 @@ func (x *scheduler) ScheduleWithCron(ctx context.Context, message proto.Message,
 		return errors.New("messages scheduler is not started")
 	}
 	// create the job
-	job := quartz.NewFunctionJob[bool](func(ctx context.Context) (bool, error) {
+	job := job.NewFunctionJob[bool](func(ctx context.Context) (bool, error) {
 		// when the job run send the message to actor
 		if err := Tell(ctx, pid, message); err != nil {
 			return false, err
@@ -222,8 +231,10 @@ func (x *scheduler) ScheduleWithCron(ctx context.Context, message proto.Message,
 		// return true when successful
 		return true, nil
 	})
+	// create the job details
+	jobDetails := quartz.NewJobDetail(job, quartz.NewJobKey(pid.ActorPath().String()))
 	// check whether the job is already scheduled
-	if err := x.distributeJobKeyOrNot(ctx, job); err != nil {
+	if err := x.distributeJobKeyOrNot(ctx, jobDetails); err != nil {
 		// skip the job scheduling when the key is already distributed
 		if errors.Is(err, errSkipJobScheduling) {
 			return nil
@@ -241,7 +252,7 @@ func (x *scheduler) ScheduleWithCron(ctx context.Context, message proto.Message,
 		return err
 	}
 	// schedule the job
-	return x.quartzScheduler.ScheduleJob(ctx, job, trigger)
+	return x.quartzScheduler.ScheduleJob(jobDetails, trigger)
 }
 
 // RemoteScheduleWithCron schedules a message to be sent to an actor in the future using a cron expression.
@@ -255,7 +266,7 @@ func (x *scheduler) RemoteScheduleWithCron(ctx context.Context, message proto.Me
 		return errors.New("messages scheduler is not started")
 	}
 	// create the job
-	job := quartz.NewFunctionJob[bool](func(ctx context.Context) (bool, error) {
+	job := job.NewFunctionJob[bool](func(ctx context.Context) (bool, error) {
 		// when the job run send the message to actor
 		if err := RemoteTell(ctx, address, message); err != nil {
 			return false, err
@@ -263,8 +274,13 @@ func (x *scheduler) RemoteScheduleWithCron(ctx context.Context, message proto.Me
 		// return true when successful
 		return true, nil
 	})
+	// create the job key name
+	key := fmt.Sprintf("%s@%s", address.GetName(), net.JoinHostPort(address.GetHost(), strconv.Itoa(int(address.GetPort()))))
+	// create the job details
+	jobDetails := quartz.NewJobDetail(job, quartz.NewJobKey(key))
+
 	// check whether the job is already scheduled
-	if err := x.distributeJobKeyOrNot(ctx, job); err != nil {
+	if err := x.distributeJobKeyOrNot(ctx, jobDetails); err != nil {
 		// skip the job scheduling when the key is already distributed
 		if errors.Is(err, errSkipJobScheduling) {
 			return nil
@@ -282,13 +298,13 @@ func (x *scheduler) RemoteScheduleWithCron(ctx context.Context, message proto.Me
 		return err
 	}
 	// schedule the job
-	return x.quartzScheduler.ScheduleJob(ctx, job, trigger)
+	return x.quartzScheduler.ScheduleJob(jobDetails, trigger)
 }
 
 // distributeJobKeyOrNot distribute the job key when it does not exist or skip it
-func (x *scheduler) distributeJobKeyOrNot(ctx context.Context, job quartz.Job) error {
+func (x *scheduler) distributeJobKeyOrNot(ctx context.Context, job *quartz.JobDetail) error {
 	// get the job key
-	jobKey := strconv.Itoa(job.Key())
+	jobKey := job.JobKey().String()
 	// check whether the cluster mode is enabled
 	if x.cluster != nil {
 		// check the job existence
diff --git a/cluster/cluster.go b/cluster/cluster.go
deleted file mode 100644
index ea9bd85..0000000
--- a/cluster/cluster.go
+++ /dev/null
@@ -1,489 +0,0 @@
-/*
- * MIT License
- *
- * Copyright (c) 2022-2024 Tochemey
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in all
- * copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
- * SOFTWARE.
- */
-
-package cluster
-
-import (
-	"context"
-	"time"
-
-	"github.com/buraksezer/olric"
-	"github.com/buraksezer/olric/config"
-	olriconfig "github.com/buraksezer/olric/config"
-	"github.com/buraksezer/olric/hasher"
-	"github.com/pkg/errors"
-	"github.com/tochemey/goakt/discovery"
-	"github.com/tochemey/goakt/hash"
-	internalpb "github.com/tochemey/goakt/internal/v1"
-	"github.com/tochemey/goakt/log"
-)
-
-// Interface defines the Cluster interface
-type Interface interface {
-	// Start starts the Cluster engine
-	Start(ctx context.Context) error
-	// Stop stops the Cluster engine
-	Stop(ctx context.Context) error
-	// NodeHost returns the Cluster node host address
-	NodeHost() string
-	// NodeRemotingPort returns the Cluster node remoting port
-	NodeRemotingPort() int
-	// PutActor replicates onto the Cluster the metadata of an actor
-	PutActor(ctx context.Context, actor *internalpb.WireActor) error
-	// GetActor fetches an actor from the Cluster
-	GetActor(ctx context.Context, actorName string) (*internalpb.WireActor, error)
-	// GetPartition returns the partition where a given actor is stored
-	GetPartition(actorName string) int
-	// SetKey sets a given key to the cluster
-	SetKey(ctx context.Context, key string) error
-	// KeyExists checks the existence of a given key
-	KeyExists(ctx context.Context, key string) (bool, error)
-	// RemoveActor removes a given actor from the cluster.
-	// An actor is removed from the cluster when this actor has been passivated.
-	RemoveActor(ctx context.Context, actorName string) error
-}
-
-// Cluster represents the Cluster
-type Cluster struct {
-	// specifies the total number of partitions
-	// the default values is 20
-	partitionsCount uint64
-
-	// specifies the logger
-	logger log.Logger
-
-	// specifies the Cluster name
-	name string
-
-	// specifies the Cluster server
-	server *olric.Olric
-	// specifies the Cluster client
-	// this help set and fetch data from the Cluster
-	client olric.Client
-
-	// specifies the distributed key value store
-	kvStore olric.DMap
-
-	// specifies the Cluster host
-	host *discovery.Node
-
-	// specifies the hasher
-	hasher hash.Hasher
-
-	// specifies the discovery provider
-	discoveryProvider discovery.Provider
-	// specifies the discovery options
-	discoveryOptions discovery.Config
-
-	writeTimeout    time.Duration
-	readTimeout     time.Duration
-	shutdownTimeout time.Duration
-}
-
-// enforce compilation error
-var _ Interface = &Cluster{}
-
-// New creates an instance of Cluster
-func New(name string, serviceDiscovery *discovery.ServiceDiscovery, opts ...Option) (*Cluster, error) {
-	// create an instance of the Cluster
-	cl := &Cluster{
-		partitionsCount:   20,
-		logger:            log.DefaultLogger,
-		name:              name,
-		discoveryProvider: serviceDiscovery.Provider(),
-		discoveryOptions:  serviceDiscovery.Config(),
-		writeTimeout:      time.Second,
-		readTimeout:       time.Second,
-		shutdownTimeout:   3 * time.Second,
-		hasher:            hash.DefaultHasher(),
-	}
-	// apply the various options
-	for _, opt := range opts {
-		opt.Apply(cl)
-	}
-
-	// get the host info
-	hostNode, err := discovery.HostNode()
-	// handle the error
-	if err != nil {
-		cl.logger.Error(errors.Wrap(err, "failed get the host node.💥"))
-		return nil, err
-	}
-
-	// set the host node
-	cl.host = hostNode
-
-	return cl, nil
-}
-
-// Start starts the Cluster.
-func (c *Cluster) Start(ctx context.Context) error {
-	// set the logger
-	logger := c.logger
-
-	// add some logging information
-	logger.Infof("Starting GoAkt Cluster service on (%s)....🤔", c.host.ClusterAddress())
-
-	// build the Cluster engine config
-	conf := c.buildConfig()
-	// set the hasher to the custom hasher
-	conf.Hasher = &hasherWrapper{c.hasher}
-
-	// create the member list config
-	m, err := olriconfig.NewMemberlistConfig("lan")
-	// panic when there is an error
-	if err != nil {
-		logger.Error(errors.Wrap(err, "failed to configure the Cluster memberlist.💥"))
-		return err
-	}
-
-	// sets the bindings
-	m.BindAddr = c.host.Host
-	m.BindPort = c.host.GossipPort
-	m.AdvertisePort = c.host.GossipPort
-	conf.MemberlistConfig = m
-
-	// set the discovery provider
-	discoveryWrapper := &discoveryProvider{
-		provider: c.discoveryProvider,
-		log:      c.logger.StdLogger(),
-	}
-	// set the discovery service
-	conf.ServiceDiscovery = map[string]any{
-		"plugin":  discoveryWrapper,
-		"id":      c.discoveryProvider.ID(),
-		"options": c.discoveryOptions,
-	}
-
-	// let us start the Cluster
-	startCtx, cancel := context.WithCancel(ctx)
-	// cancel the context the server has started
-	conf.Started = func() {
-		// cancel the start context
-		defer cancel()
-		// add some logging information
-		logger.Info("GoAkt Cluster successfully started. 🎉")
-	}
-
-	// let us create an instance of the Cluster engine
-	eng, err := olric.New(conf)
-	// handle the error
-	if err != nil {
-		logger.Error(errors.Wrap(err, "failed to start the Cluster engine.💥"))
-		return err
-	}
-
-	// set the server
-	c.server = eng
-	go func() {
-		// start the Cluster engine
-		err = c.server.Start()
-		// handle the error in case there is an early error
-		if err != nil {
-			logger.Error(errors.Wrap(err, "failed to start the Cluster engine.💥"))
-			// let us stop the started engine
-			if e := c.server.Shutdown(ctx); e != nil {
-				logger.Panic(e)
-			}
-		}
-	}()
-
-	// wait for start
-	<-startCtx.Done()
-
-	// set the client
-	c.client = c.server.NewEmbeddedClient()
-	// create the instance of the distributed map
-	dmp, err := c.client.NewDMap(c.name)
-	// handle the error
-	if err != nil {
-		logger.Error(errors.Wrap(err, "failed to start the Cluster engine.💥"))
-		// let us stop the started engine
-		return c.server.Shutdown(ctx)
-	}
-
-	// set the distributed map
-	c.kvStore = dmp
-	return nil
-}
-
-// Stop stops the Cluster gracefully
-func (c *Cluster) Stop(ctx context.Context) error {
-	// create a cancellation context of 1 second timeout
-	ctx, cancelFn := context.WithTimeout(ctx, c.shutdownTimeout)
-	defer cancelFn()
-
-	// set the logger
-	logger := c.logger
-
-	// add some logging information
-	logger.Info("Stopping GoAkt Cluster....🤔")
-
-	// close the Cluster client
-	if err := c.client.Close(ctx); err != nil {
-		logger.Error(errors.Wrap(err, "failed to shutdown the Cluster client.💥"))
-		return err
-	}
-
-	// let us stop the server
-	if err := c.server.Shutdown(ctx); err != nil {
-		logger.Error(errors.Wrap(err, "failed to Shutdown  GoAkt Cluster....💥"))
-		return err
-	}
-
-	logger.Info("GoAkt Cluster successfully stopped.🎉")
-	return nil
-}
-
-// NodeHost returns the Cluster node Host
-func (c *Cluster) NodeHost() string {
-	return c.host.Host
-}
-
-// NodeRemotingPort returns the Cluster node remoting port
-func (c *Cluster) NodeRemotingPort() int {
-	return c.host.RemotingPort
-}
-
-// PutActor replicates onto the Cluster the metadata of an actor
-func (c *Cluster) PutActor(ctx context.Context, actor *internalpb.WireActor) error {
-	// create a cancellation context of 1 second timeout
-	ctx, cancelFn := context.WithTimeout(ctx, c.writeTimeout)
-	defer cancelFn()
-
-	// set the logger
-	logger := c.logger
-
-	// add some logging information
-	logger.Infof("replicating actor (%s).🤔", actor.GetActorName())
-
-	// let us marshal it
-	data, err := encode(actor)
-	// handle the marshaling error
-	if err != nil {
-		// add a logging to the stderr
-		logger.Error(errors.Wrapf(err, "failed to persist actor=%s data in the Cluster.💥", actor.GetActorName()))
-		// here we cancel the request
-		return errors.Wrapf(err, "failed to persist actor=%s data in the Cluster", actor.GetActorName())
-	}
-
-	// send the record into the Cluster
-	err = c.kvStore.Put(ctx, actor.GetActorName(), data)
-	// handle the error
-	if err != nil {
-		// log the error
-		logger.Error(errors.Wrapf(err, "failed to replicate actor=%s record.💥", actor.GetActorName()))
-		return err
-	}
-
-	// Ahoy we are successful
-	logger.Infof("actor (%s) successfully replicated.🎉", actor.GetActorName())
-	return nil
-}
-
-// GetActor fetches an actor from the Cluster
-func (c *Cluster) GetActor(ctx context.Context, actorName string) (*internalpb.WireActor, error) {
-	// create a cancellation context of 1 second timeout
-	ctx, cancelFn := context.WithTimeout(ctx, c.readTimeout)
-	defer cancelFn()
-
-	// set the logger
-	logger := c.logger
-
-	// add some logging information
-	logger.Infof("retrieving actor (%s) from the Cluster.🤔", actorName)
-
-	// grab the record from the distributed store
-	resp, err := c.kvStore.Get(ctx, actorName)
-	// handle the error
-	if err != nil {
-		// we could not find the given actor
-		if errors.Is(err, olric.ErrKeyNotFound) {
-			logger.Warnf("actor=%s is not found in the Cluster", actorName)
-			return nil, ErrActorNotFound
-		}
-		// log the error
-		logger.Error(errors.Wrapf(err, "failed to get actor=%s record.💥", actorName))
-		return nil, err
-	}
-
-	// grab the base64 representation of the wire actor
-	base64ActorStr, err := resp.String()
-	// handle the error
-	if err != nil {
-		logger.Error(errors.Wrapf(err, "failed to read the record at:{%s}.💥", actorName))
-		return nil, err
-	}
-
-	// decode it
-	actor, err := decode(base64ActorStr)
-	// let us unpack the byte array
-	if err != nil {
-		// log the error and return
-		logger.Error(errors.Wrapf(err, "failed to decode actor=%s record.💥", actorName))
-		return nil, err
-	}
-
-	// Ahoy we are successful
-	logger.Infof("actor (%s) successfully retrieved from the cluster.🎉", actor.GetActorName())
-	// return the response
-	return actor, nil
-}
-
-// RemoveActor removes a given actor from the cluster.
-// An actor is removed from the cluster when this actor has been passivated.
-func (c *Cluster) RemoveActor(ctx context.Context, actorName string) error {
-	// set the logger
-	logger := c.logger
-
-	// add some logging information
-	logger.Infof("removing actor (%s).🤔", actorName)
-
-	// remove the actor from the cluster
-	_, err := c.kvStore.Delete(ctx, actorName)
-	// handle the error
-	if err != nil {
-		// log the error
-		logger.Error(errors.Wrapf(err, "failed to remove actor=%s record.💥", actorName))
-		return err
-	}
-
-	// add a logging information
-	logger.Infof("actor (%s) successfully removed from the cluster.🎉", actorName)
-	return nil
-}
-
-// SetKey sets a given key to the cluster
-func (c *Cluster) SetKey(ctx context.Context, key string) error {
-	// create a cancellation context of 1 second timeout
-	ctx, cancelFn := context.WithTimeout(ctx, c.writeTimeout)
-	defer cancelFn()
-
-	// set the logger
-	logger := c.logger
-
-	// add some logging information
-	logger.Infof("replicating key (%s).🤔", key)
-
-	// send the record into the Cluster
-	err := c.kvStore.Put(ctx, key, true)
-	// handle the error
-	if err != nil {
-		// log the error
-		logger.Error(errors.Wrapf(err, "failed to replicate key=%s record.💥", key))
-		return err
-	}
-
-	// Ahoy we are successful
-	logger.Infof("key (%s) successfully replicated.🎉", key)
-	return nil
-}
-
-// KeyExists checks the existence of a given key
-func (c *Cluster) KeyExists(ctx context.Context, key string) (bool, error) {
-	// create a cancellation context of 1 second timeout
-	ctx, cancelFn := context.WithTimeout(ctx, c.readTimeout)
-	defer cancelFn()
-
-	// set the logger
-	logger := c.logger
-
-	// add some logging information
-	logger.Infof("checking key (%s) existence in the cluster.🤔", key)
-
-	// grab the record from the distributed store
-	resp, err := c.kvStore.Get(ctx, key)
-	// handle the error
-	if err != nil {
-		// we could not find the given actor
-		if errors.Is(err, olric.ErrKeyNotFound) {
-			logger.Warnf("key=%s is not found in the Cluster", key)
-			return false, nil
-		}
-		// log the error
-		logger.Error(errors.Wrapf(err, "failed to check key=%s existence.💥", key))
-		return false, err
-	}
-	return resp.Bool()
-}
-
-// GetPartition returns the partition where a given actor is stored
-func (c *Cluster) GetPartition(actorName string) int {
-	// create the byte array of the actor name
-	key := []byte(actorName)
-	// compute the hash key
-	hkey := c.hasher.HashCode(key)
-	// compute the partition and return it
-	partition := int(hkey % c.partitionsCount)
-	// add some debug log
-	c.logger.Debugf("partition of actor (%s) is (%d)", actorName, partition)
-	return partition
-}
-
-// buildConfig builds the Cluster configuration
-func (c *Cluster) buildConfig() *config.Config {
-	// define the log level
-	logLevel := "INFO"
-	switch c.logger.LogLevel() {
-	case log.DebugLevel:
-		logLevel = "DEBUG"
-	case log.ErrorLevel, log.FatalLevel, log.PanicLevel:
-		logLevel = "ERROR"
-	case log.WarningLevel:
-		logLevel = "WARN"
-	default:
-		// pass
-	}
-
-	// create the config and return it
-	conf := &config.Config{
-		BindAddr:                   c.host.Host,
-		BindPort:                   c.host.ClusterPort,
-		ReadRepair:                 false,
-		ReplicaCount:               config.MinimumReplicaCount,
-		WriteQuorum:                config.DefaultWriteQuorum,
-		ReadQuorum:                 config.DefaultReadQuorum,
-		MemberCountQuorum:          config.DefaultMemberCountQuorum,
-		Peers:                      []string{},
-		DMaps:                      &olriconfig.DMaps{},
-		KeepAlivePeriod:            config.DefaultKeepAlivePeriod,
-		PartitionCount:             c.partitionsCount,
-		BootstrapTimeout:           config.DefaultBootstrapTimeout,
-		ReplicationMode:            olriconfig.SyncReplicationMode,
-		RoutingTablePushInterval:   config.DefaultRoutingTablePushInterval,
-		JoinRetryInterval:          config.DefaultJoinRetryInterval,
-		MaxJoinAttempts:            config.DefaultMaxJoinAttempts,
-		LogLevel:                   logLevel,
-		LogOutput:                  newLogWriter(c.logger),
-		EnableClusterEventsChannel: false,
-		Hasher:                     hasher.NewDefaultHasher(),
-	}
-
-	// set verbosity when debug is enabled
-	if c.logger.LogLevel() == log.DebugLevel {
-		conf.LogVerbosity = config.DefaultLogVerbosity
-	}
-
-	return conf
-}
diff --git a/cluster/node.go b/cluster/node.go
new file mode 100644
index 0000000..c42d580
--- /dev/null
+++ b/cluster/node.go
@@ -0,0 +1,649 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2022-2024 Tochemey
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package cluster
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/buraksezer/olric"
+	"github.com/buraksezer/olric/config"
+	olriconfig "github.com/buraksezer/olric/config"
+	"github.com/buraksezer/olric/events"
+	"github.com/buraksezer/olric/hasher"
+	"github.com/go-redis/redis/v8"
+	"github.com/pkg/errors"
+	"github.com/tochemey/goakt/discovery"
+	"github.com/tochemey/goakt/hash"
+	internalpb "github.com/tochemey/goakt/internal/v1"
+	"github.com/tochemey/goakt/log"
+	eventspb "github.com/tochemey/goakt/pb/events/v1"
+	"github.com/tochemey/goakt/pkg/types"
+	"google.golang.org/protobuf/types/known/anypb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+// Event defines the cluster event
+type Event struct {
+	Type *anypb.Any
+}
+
+// Interface defines the Node interface
+type Interface interface {
+	// Start starts the Node engine
+	Start(ctx context.Context) error
+	// Stop stops the Node engine
+	Stop(ctx context.Context) error
+	// NodeHost returns the cluster startNode host address
+	NodeHost() string
+	// NodeRemotingPort returns the cluster startNode remoting port
+	NodeRemotingPort() int
+	// PutActor replicates onto the Node the metadata of an actor
+	PutActor(ctx context.Context, actor *internalpb.WireActor) error
+	// GetActor fetches an actor from the Node
+	GetActor(ctx context.Context, actorName string) (*internalpb.WireActor, error)
+	// GetPartition returns the partition where a given actor is stored
+	GetPartition(actorName string) int
+	// SetKey sets a given key to the cluster
+	SetKey(ctx context.Context, key string) error
+	// KeyExists checks the existence of a given key
+	KeyExists(ctx context.Context, key string) (bool, error)
+	// RemoveActor removes a given actor from the cluster.
+	// An actor is removed from the cluster when this actor has been passivated.
+	RemoveActor(ctx context.Context, actorName string) error
+	// Events returns a channel where cluster events are published
+	Events() <-chan *Event
+	// AdvertisedAddress returns the cluster node cluster address that is known by the
+	// peers in the cluster
+	AdvertisedAddress() string
+}
+
+// Node represents the Node
+type Node struct {
+	// specifies the total number of partitions
+	// the default values is 20
+	partitionsCount uint64
+
+	// specifies the logger
+	logger log.Logger
+
+	// specifies the Node name
+	name string
+
+	// specifies the Node server
+	server *olric.Olric
+	// specifies the Node client
+	// this help set and fetch data from the Node
+	client olric.Client
+
+	// specifies the distributed key value store
+	kvStore olric.DMap
+
+	// specifies the Node host
+	host *discovery.Node
+
+	// specifies the hasher
+	hasher hash.Hasher
+
+	// specifies the discovery provider
+	discoveryProvider discovery.Provider
+	// specifies the discovery options
+	discoveryOptions discovery.Config
+
+	writeTimeout    time.Duration
+	readTimeout     time.Duration
+	shutdownTimeout time.Duration
+
+	events             chan *Event
+	pubSub             *redis.PubSub
+	messagesChan       <-chan *redis.Message
+	messagesReaderChan chan types.Unit
+}
+
+// enforce compilation error
+var _ Interface = &Node{}
+
+// NewNode creates an instance of cluster Node
+func NewNode(name string, serviceDiscovery *discovery.ServiceDiscovery, opts ...Option) (*Node, error) {
+	// create an instance of the Node
+	node := &Node{
+		partitionsCount:    20,
+		logger:             log.DefaultLogger,
+		name:               name,
+		discoveryProvider:  serviceDiscovery.Provider(),
+		discoveryOptions:   serviceDiscovery.Config(),
+		writeTimeout:       time.Second,
+		readTimeout:        time.Second,
+		shutdownTimeout:    3 * time.Second,
+		hasher:             hash.DefaultHasher(),
+		pubSub:             nil,
+		events:             make(chan *Event, 20),
+		messagesReaderChan: make(chan types.Unit, 1),
+		messagesChan:       make(chan *redis.Message, 1),
+	}
+	// apply the various options
+	for _, opt := range opts {
+		opt.Apply(node)
+	}
+
+	// get the host info
+	hostNode, err := discovery.HostNode()
+	// handle the error
+	if err != nil {
+		node.logger.Error(errors.Wrap(err, "failed get the host node.💥"))
+		return nil, err
+	}
+
+	// set the host startNode
+	node.host = hostNode
+
+	return node, nil
+}
+
+// Start starts the Node.
+func (n *Node) Start(ctx context.Context) error {
+	// set the logger
+	logger := n.logger
+
+	// add some logging information
+	logger.Infof("Starting GoAkt cluster Node service on (%s)....🤔", n.host.ClusterAddress())
+
+	// build the Node engine config
+	conf := n.buildConfig()
+	// set the hasher to the custom hasher
+	conf.Hasher = &hasherWrapper{n.hasher}
+
+	// create the member list config
+	m, err := olriconfig.NewMemberlistConfig("lan")
+	// panic when there is an error
+	if err != nil {
+		logger.Error(errors.Wrap(err, "failed to configure the cluster Node memberlist.💥"))
+		return err
+	}
+
+	// sets the bindings
+	m.BindAddr = n.host.Host
+	m.BindPort = n.host.GossipPort
+	m.AdvertisePort = n.host.GossipPort
+	conf.MemberlistConfig = m
+
+	// set the discovery provider
+	discoveryWrapper := &discoveryProvider{
+		provider: n.discoveryProvider,
+		log:      n.logger.StdLogger(),
+	}
+	// set the discovery service
+	conf.ServiceDiscovery = map[string]any{
+		"plugin":  discoveryWrapper,
+		"id":      n.discoveryProvider.ID(),
+		"options": n.discoveryOptions,
+	}
+
+	// let us start the Node
+	startCtx, cancel := context.WithCancel(ctx)
+	// cancel the context the server has started
+	conf.Started = func() {
+		// cancel the start context
+		defer cancel()
+		// add some logging information
+		logger.Infof("GoAkt cluster Node=(%s) successfully started. 🎉", n.name)
+	}
+
+	// let us create an instance of the Node engine
+	eng, err := olric.New(conf)
+	// handle the error
+	if err != nil {
+		logger.Error(errors.Wrapf(err, "failed to start the cluster Node=(%s).💥", n.name))
+		return err
+	}
+
+	// set the server
+	n.server = eng
+	go func() {
+		// start the Node engine
+		err = n.server.Start()
+		// handle the error in case there is an early error
+		if err != nil {
+			logger.Error(errors.Wrapf(err, "failed to start the cluster Node=(%s).💥", n.name))
+			// let us stop the started engine
+			if e := n.server.Shutdown(ctx); e != nil {
+				logger.Panic(e)
+			}
+		}
+	}()
+
+	// wait for start
+	<-startCtx.Done()
+
+	// set the client
+	n.client = n.server.NewEmbeddedClient()
+	// create the instance of the distributed map
+	dmp, err := n.client.NewDMap(n.name)
+	// handle the error
+	if err != nil {
+		logger.Error(errors.Wrapf(err, "failed to start the cluster Node=(%s).💥", n.name))
+		// let us stop the started engine
+		return n.server.Shutdown(ctx)
+	}
+	// set the distributed map
+	n.kvStore = dmp
+
+	// create a subscriber to consume to cluster events
+	ps, err := n.client.NewPubSub()
+	// handle the error
+	if err != nil {
+		logger.Error(errors.Wrapf(err, "failed to start the cluster Node=(%s).💥", n.name))
+		// let us stop the started engine
+		return n.server.Shutdown(ctx)
+	}
+
+	// subscribe to the cluster events channel
+	n.pubSub = ps.Subscribe(ctx, events.ClusterEventsChannel)
+	// set the events channel
+	n.messagesChan = n.pubSub.Channel()
+	// start consuming to cluster events
+	go n.consume()
+	// return successfully
+	return nil
+}
+
+// Stop stops the Node gracefully
+func (n *Node) Stop(ctx context.Context) error {
+	// create a cancellation context of 1 second timeout
+	ctx, cancelFn := context.WithTimeout(ctx, n.shutdownTimeout)
+	defer cancelFn()
+
+	// set the logger
+	logger := n.logger
+
+	// add some logging information
+	logger.Infof("Stopping GoAkt cluster Node=(%s)....🤔", n.name)
+
+	// close the events listener
+	if err := n.pubSub.Close(); err != nil {
+		logger.Error(errors.Wrap(err, "failed to shutdown the cluster events listener.💥"))
+		return err
+	}
+
+	// close the Node client
+	if err := n.client.Close(ctx); err != nil {
+		logger.Error(errors.Wrapf(err, "failed to shutdown the cluster Node=(%s).💥", n.name))
+		return err
+	}
+
+	// let us stop the server
+	if err := n.server.Shutdown(ctx); err != nil {
+		logger.Error(errors.Wrapf(err, "failed to Shutdown GoAkt cluster Node=(%s)....💥", n.name))
+		return err
+	}
+
+	// close the events queue
+	close(n.events)
+
+	// signal we are stopping listening to events
+	n.messagesReaderChan <- types.Unit{}
+
+	logger.Infof("GoAkt cluster Node=(%s) successfully stopped.🎉", n.name)
+	return nil
+}
+
+// NodeHost returns the Node Host
+func (n *Node) NodeHost() string {
+	return n.host.Host
+}
+
+// NodeRemotingPort returns the Node remoting port
+func (n *Node) NodeRemotingPort() int {
+	return n.host.RemotingPort
+}
+
+// AdvertisedAddress returns the cluster node cluster address that is known by the
+// peers in the cluster
+func (n *Node) AdvertisedAddress() string {
+	return n.host.ClusterAddress()
+}
+
+// PutActor replicates onto the Node the metadata of an actor
+func (n *Node) PutActor(ctx context.Context, actor *internalpb.WireActor) error {
+	// create a cancellation context of 1 second timeout
+	ctx, cancelFn := context.WithTimeout(ctx, n.writeTimeout)
+	defer cancelFn()
+
+	// set the logger
+	logger := n.logger
+
+	// add some logging information
+	logger.Infof("replicating actor (%s).🤔", actor.GetActorName())
+
+	// let us marshal it
+	data, err := encode(actor)
+	// handle the marshaling error
+	if err != nil {
+		// add a logging to the stderr
+		logger.Error(errors.Wrapf(err, "failed to persist actor=%s data in the Node.💥", actor.GetActorName()))
+		// here we cancel the request
+		return errors.Wrapf(err, "failed to persist actor=%s data in the Node", actor.GetActorName())
+	}
+
+	// send the record into the Node
+	err = n.kvStore.Put(ctx, actor.GetActorName(), data)
+	// handle the error
+	if err != nil {
+		// log the error
+		logger.Error(errors.Wrapf(err, "failed to replicate actor=%s record.💥", actor.GetActorName()))
+		return err
+	}
+
+	// Ahoy we are successful
+	logger.Infof("actor (%s) successfully replicated.🎉", actor.GetActorName())
+	return nil
+}
+
+// GetActor fetches an actor from the Node
+func (n *Node) GetActor(ctx context.Context, actorName string) (*internalpb.WireActor, error) {
+	// create a cancellation context of 1 second timeout
+	ctx, cancelFn := context.WithTimeout(ctx, n.readTimeout)
+	defer cancelFn()
+
+	// set the logger
+	logger := n.logger
+
+	// add some logging information
+	logger.Infof("retrieving actor (%s) from the Node.🤔", actorName)
+
+	// grab the record from the distributed store
+	resp, err := n.kvStore.Get(ctx, actorName)
+	// handle the error
+	if err != nil {
+		// we could not find the given actor
+		if errors.Is(err, olric.ErrKeyNotFound) {
+			logger.Warnf("actor=%s is not found in the Node", actorName)
+			return nil, ErrActorNotFound
+		}
+		// log the error
+		logger.Error(errors.Wrapf(err, "failed to get actor=%s record.💥", actorName))
+		return nil, err
+	}
+
+	// grab the base64 representation of the wire actor
+	base64ActorStr, err := resp.String()
+	// handle the error
+	if err != nil {
+		logger.Error(errors.Wrapf(err, "failed to read the record at:{%s}.💥", actorName))
+		return nil, err
+	}
+
+	// decode it
+	actor, err := decode(base64ActorStr)
+	// let us unpack the byte array
+	if err != nil {
+		// log the error and return
+		logger.Error(errors.Wrapf(err, "failed to decode actor=%s record.💥", actorName))
+		return nil, err
+	}
+
+	// Ahoy we are successful
+	logger.Infof("actor (%s) successfully retrieved from the cluster.🎉", actor.GetActorName())
+	// return the response
+	return actor, nil
+}
+
+// RemoveActor removes a given actor from the cluster.
+// An actor is removed from the cluster when this actor has been passivated.
+func (n *Node) RemoveActor(ctx context.Context, actorName string) error {
+	// set the logger
+	logger := n.logger
+
+	// add some logging information
+	logger.Infof("removing actor (%s).🤔", actorName)
+
+	// remove the actor from the cluster
+	_, err := n.kvStore.Delete(ctx, actorName)
+	// handle the error
+	if err != nil {
+		// log the error
+		logger.Error(errors.Wrapf(err, "failed to remove actor=%s record.💥", actorName))
+		return err
+	}
+
+	// add a logging information
+	logger.Infof("actor (%s) successfully removed from the cluster.🎉", actorName)
+	return nil
+}
+
+// SetKey sets a given key to the cluster
+func (n *Node) SetKey(ctx context.Context, key string) error {
+	// create a cancellation context of 1 second timeout
+	ctx, cancelFn := context.WithTimeout(ctx, n.writeTimeout)
+	defer cancelFn()
+
+	// set the logger
+	logger := n.logger
+
+	// add some logging information
+	logger.Infof("replicating key (%s).🤔", key)
+
+	// send the record into the Node
+	err := n.kvStore.Put(ctx, key, true)
+	// handle the error
+	if err != nil {
+		// log the error
+		logger.Error(errors.Wrapf(err, "failed to replicate key=%s record.💥", key))
+		return err
+	}
+
+	// Ahoy we are successful
+	logger.Infof("key (%s) successfully replicated.🎉", key)
+	return nil
+}
+
+// KeyExists checks the existence of a given key
+func (n *Node) KeyExists(ctx context.Context, key string) (bool, error) {
+	// create a cancellation context of 1 second timeout
+	ctx, cancelFn := context.WithTimeout(ctx, n.readTimeout)
+	defer cancelFn()
+
+	// set the logger
+	logger := n.logger
+
+	// add some logging information
+	logger.Infof("checking key (%s) existence in the cluster.🤔", key)
+
+	// grab the record from the distributed store
+	resp, err := n.kvStore.Get(ctx, key)
+	// handle the error
+	if err != nil {
+		// we could not find the given actor
+		if errors.Is(err, olric.ErrKeyNotFound) {
+			logger.Warnf("key=%s is not found in the Node", key)
+			return false, nil
+		}
+		// log the error
+		logger.Error(errors.Wrapf(err, "failed to check key=%s existence.💥", key))
+		return false, err
+	}
+	return resp.Bool()
+}
+
+// GetPartition returns the partition where a given actor is stored
+func (n *Node) GetPartition(actorName string) int {
+	// create the byte array of the actor name
+	key := []byte(actorName)
+	// compute the hash key
+	hkey := n.hasher.HashCode(key)
+	// compute the partition and return it
+	partition := int(hkey % n.partitionsCount)
+	// add some debug log
+	n.logger.Debugf("partition of actor (%s) is (%d)", actorName, partition)
+	return partition
+}
+
+// Events returns a channel where cluster events are published
+func (n *Node) Events() <-chan *Event {
+	return n.events
+}
+
+// consume reads to the underlying cluster events
+// and emit the event
+func (n *Node) consume() {
+	for {
+		select {
+		case <-n.messagesReaderChan:
+			return
+		case message, ok := <-n.messagesChan:
+			// break out of the loop when the channel is closed
+			if !ok {
+				return
+			}
+			// grab the message payload
+			payload := message.Payload
+			// get the message payload
+			var event map[string]any
+			// let us unmarshal the event
+			if err := json.Unmarshal([]byte(payload), &event); err != nil {
+				n.logger.Error(errors.Wrap(err, "failed to decode cluster event"))
+				// TODO: should we continue or not
+				continue
+			}
+
+			// grab the kind
+			kind := event["kind"]
+			// add some debug log
+			n.logger.Debugf("%s received (%s) cluster event", n.name, kind)
+
+			switch kind {
+			case events.KindNodeJoinEvent:
+				// create the node joined to unmarshal the event
+				nodeJoined := new(events.NodeJoinEvent)
+				// let us unmarshal the event
+				if err := json.Unmarshal([]byte(payload), &nodeJoined); err != nil {
+					n.logger.Error(errors.Wrap(err, "failed to decode NodeJoined cluster event"))
+					// TODO: should we continue or not
+					continue
+				}
+
+				// make sure to skip self
+				if n.AdvertisedAddress() == nodeJoined.NodeJoin {
+					// add some debug log
+					n.logger.Debug("skipping self")
+					continue
+				}
+
+				// TODO: need to cross check this calculation
+				// convert the timestamp to milliseconds
+				timeMilli := nodeJoined.Timestamp / int64(1e6)
+				// create the startNode joined event
+				event := &eventspb.NodeJoined{
+					Address:   nodeJoined.NodeJoin,
+					Timestamp: timestamppb.New(time.UnixMilli(timeMilli)),
+				}
+				// serialize as any pb
+				eventType, _ := anypb.New(event)
+				// send the event to queue
+				n.events <- &Event{eventType}
+
+			case events.KindNodeLeftEvent:
+				// create the node left to unmarshal the event
+				nodeLeft := new(events.NodeLeftEvent)
+				// let us unmarshal the event
+				if err := json.Unmarshal([]byte(payload), &nodeLeft); err != nil {
+					n.logger.Error(errors.Wrap(err, "failed to decode NodeLeft cluster event"))
+					// TODO: should we continue or not
+					continue
+				}
+
+				// make sure to skip self
+				if n.AdvertisedAddress() == nodeLeft.NodeLeft {
+					// add some debug log
+					n.logger.Debug("skipping self")
+					continue
+				}
+
+				// TODO: need to cross check this calculation
+				// convert the timestamp to milliseconds
+				timeMilli := nodeLeft.Timestamp / int64(1e6)
+				// create the startNode joined event
+				event := &eventspb.NodeLeft{
+					Address:   nodeLeft.NodeLeft,
+					Timestamp: timestamppb.New(time.UnixMilli(timeMilli)),
+				}
+				// serialize as any pb
+				eventType, _ := anypb.New(event)
+				// send the event to queue
+				n.events <- &Event{eventType}
+			default:
+				// skip
+			}
+		}
+	}
+}
+
+// buildConfig builds the Node configuration
+func (n *Node) buildConfig() *config.Config {
+	// define the log level
+	logLevel := "INFO"
+	switch n.logger.LogLevel() {
+	case log.DebugLevel:
+		logLevel = "DEBUG"
+	case log.ErrorLevel, log.FatalLevel, log.PanicLevel:
+		logLevel = "ERROR"
+	case log.WarningLevel:
+		logLevel = "WARN"
+	default:
+		// pass
+	}
+
+	// create the config and return it
+	conf := &config.Config{
+		BindAddr:                   n.host.Host,
+		BindPort:                   n.host.ClusterPort,
+		ReadRepair:                 false,
+		ReplicaCount:               config.MinimumReplicaCount,
+		WriteQuorum:                config.DefaultWriteQuorum,
+		ReadQuorum:                 config.DefaultReadQuorum,
+		MemberCountQuorum:          config.DefaultMemberCountQuorum,
+		Peers:                      []string{},
+		DMaps:                      &olriconfig.DMaps{},
+		KeepAlivePeriod:            config.DefaultKeepAlivePeriod,
+		PartitionCount:             n.partitionsCount,
+		BootstrapTimeout:           config.DefaultBootstrapTimeout,
+		ReplicationMode:            olriconfig.SyncReplicationMode,
+		RoutingTablePushInterval:   config.DefaultRoutingTablePushInterval,
+		JoinRetryInterval:          config.DefaultJoinRetryInterval,
+		MaxJoinAttempts:            config.DefaultMaxJoinAttempts,
+		LogLevel:                   logLevel,
+		LogOutput:                  newLogWriter(n.logger),
+		EnableClusterEventsChannel: true,
+		Hasher:                     hasher.NewDefaultHasher(),
+	}
+
+	// set verbosity when debug is enabled
+	if n.logger.LogLevel() == log.DebugLevel {
+		conf.LogVerbosity = config.DefaultLogVerbosity
+	}
+
+	return conf
+}
diff --git a/cluster/option.go b/cluster/option.go
index f0a0cd1..8b9ba59 100644
--- a/cluster/option.go
+++ b/cluster/option.go
@@ -34,59 +34,59 @@ import (
 // Option is the interface that applies a configuration option.
 type Option interface {
 	// Apply sets the Option value of a config.
-	Apply(cl *Cluster)
+	Apply(cl *Node)
 }
 
 var _ Option = OptionFunc(nil)
 
 // OptionFunc implements the Option interface.
-type OptionFunc func(cl *Cluster)
+type OptionFunc func(cl *Node)
 
-// Apply applies the Cluster's option
-func (f OptionFunc) Apply(c *Cluster) {
+// Apply applies the Node's option
+func (f OptionFunc) Apply(c *Node) {
 	f(c)
 }
 
 // WithPartitionsCount sets the total number of partitions
 func WithPartitionsCount(count uint64) Option {
-	return OptionFunc(func(cl *Cluster) {
+	return OptionFunc(func(cl *Node) {
 		cl.partitionsCount = count
 	})
 }
 
 // WithLogger sets the logger
 func WithLogger(logger log.Logger) Option {
-	return OptionFunc(func(cl *Cluster) {
+	return OptionFunc(func(cl *Node) {
 		cl.logger = logger
 	})
 }
 
-// WithWriteTimeout sets the Cluster write timeout.
+// WithWriteTimeout sets the Node write timeout.
 // This timeout specifies the timeout of a data replication
 func WithWriteTimeout(timeout time.Duration) Option {
-	return OptionFunc(func(cl *Cluster) {
+	return OptionFunc(func(cl *Node) {
 		cl.writeTimeout = timeout
 	})
 }
 
-// WithReadTimeout sets the Cluster read timeout.
+// WithReadTimeout sets the Node read timeout.
 // This timeout specifies the timeout of a data retrieval
 func WithReadTimeout(timeout time.Duration) Option {
-	return OptionFunc(func(cl *Cluster) {
+	return OptionFunc(func(cl *Node) {
 		cl.readTimeout = timeout
 	})
 }
 
-// WithShutdownTimeout sets the Cluster shutdown timeout.
+// WithShutdownTimeout sets the Node shutdown timeout.
 func WithShutdownTimeout(timeout time.Duration) Option {
-	return OptionFunc(func(cl *Cluster) {
+	return OptionFunc(func(cl *Node) {
 		cl.shutdownTimeout = timeout
 	})
 }
 
 // WithHasher sets the custom hasher
 func WithHasher(hasher hash.Hasher) Option {
-	return OptionFunc(func(cl *Cluster) {
+	return OptionFunc(func(cl *Node) {
 		cl.hasher = hasher
 	})
 }
diff --git a/go.mod b/go.mod
index 8c66f78..f05e5af 100644
--- a/go.mod
+++ b/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/deckarep/golang-set/v2 v2.6.0
 	github.com/flowchartsman/retry v1.2.0
+	github.com/go-redis/redis/v8 v8.11.5
 	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.5.0
 	github.com/grandcat/zeroconf v1.0.0
@@ -18,7 +19,7 @@ require (
 	github.com/nats-io/nats.go v1.32.0
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.18.0
-	github.com/reugn/go-quartz v0.9.0
+	github.com/reugn/go-quartz v0.10.0
 	github.com/spf13/cobra v1.8.0
 	github.com/stretchr/testify v1.8.4
 	github.com/travisjeffery/go-dynaport v1.0.0
@@ -35,9 +36,9 @@ require (
 	go.uber.org/zap v1.26.0
 	golang.org/x/net v0.20.0
 	google.golang.org/protobuf v1.32.0
-	k8s.io/api v0.29.0
-	k8s.io/apimachinery v0.29.0
-	k8s.io/client-go v0.29.0
+	k8s.io/api v0.29.1
+	k8s.io/apimachinery v0.29.1
+	k8s.io/client-go v0.29.1
 	k8s.io/utils v0.0.0-20240102154912-e7106e64919e
 )
 
@@ -58,7 +59,6 @@ require (
 	github.com/go-openapi/jsonpointer v0.20.0 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.22.4 // indirect
-	github.com/go-redis/redis/v8 v8.11.5 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/google/btree v1.1.2 // indirect
@@ -105,14 +105,14 @@ require (
 	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/crypto v0.18.0 // indirect
-	golang.org/x/mod v0.12.0 // indirect
+	golang.org/x/mod v0.14.0 // indirect
 	golang.org/x/oauth2 v0.12.0 // indirect
-	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/sync v0.5.0 // indirect
 	golang.org/x/sys v0.16.0 // indirect
 	golang.org/x/term v0.16.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
-	golang.org/x/tools v0.13.0 // indirect
+	golang.org/x/tools v0.16.1 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect
diff --git a/go.sum b/go.sum
index a5ffd3e..84d0110 100644
--- a/go.sum
+++ b/go.sum
@@ -263,8 +263,8 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
 github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
 github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
-github.com/reugn/go-quartz v0.9.0 h1:JPaxyWx6YtNG3020DuT/LXRqishMokWwpMiLfiTWGJg=
-github.com/reugn/go-quartz v0.9.0/go.mod h1:no4ktgYbAAuY0E1SchR8cTx1LF4jYIzdgaQhzRPSkpk=
+github.com/reugn/go-quartz v0.10.0 h1:pOFJDOjOD/RpoCHyvx6IwPLU2A9vGquyPwQuTxdvjow=
+github.com/reugn/go-quartz v0.10.0/go.mod h1:no4ktgYbAAuY0E1SchR8cTx1LF4jYIzdgaQhzRPSkpk=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -351,8 +351,8 @@ golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzB
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.12.0 h1:rmsUpXtvNzj340zd98LZ4KntptpfRHwpFOHG188oHXc=
-golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/mod v0.14.0 h1:dGoOF9QVLYng8IHTm7BAyWqCqSheQ5pYWGhzW00YJr0=
+golang.org/x/mod v0.14.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
@@ -378,8 +378,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
-golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
+golang.org/x/sync v0.5.0 h1:60k92dhOjHxJkrqnwsfl8KuaHbn/5dl0lUPUklKo3qE=
+golang.org/x/sync v0.5.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -426,8 +426,8 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.6-0.20210726203631-07bc1bf47fb2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
-golang.org/x/tools v0.13.0 h1:Iey4qkscZuv0VvIt8E0neZjtPVQFSc870HQ448QgEmQ=
-golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58=
+golang.org/x/tools v0.16.1 h1:TLyB3WofjdOEepBHAU20JdNC1Zbg87elYofWYAY5oZA=
+golang.org/x/tools v0.16.1/go.mod h1:kYVVN6I1mBNoB1OX+noeBjbRk4IUEPa7JJ+TJMEooJ0=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 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=
@@ -473,12 +473,12 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 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=
-k8s.io/api v0.29.0 h1:NiCdQMY1QOp1H8lfRyeEf8eOwV6+0xA6XEE44ohDX2A=
-k8s.io/api v0.29.0/go.mod h1:sdVmXoz2Bo/cb77Pxi71IPTSErEW32xa4aXwKH7gfBA=
-k8s.io/apimachinery v0.29.0 h1:+ACVktwyicPz0oc6MTMLwa2Pw3ouLAfAon1wPLtG48o=
-k8s.io/apimachinery v0.29.0/go.mod h1:eVBxQ/cwiJxH58eK/jd/vAk4mrxmVlnpBH5J2GbMeis=
-k8s.io/client-go v0.29.0 h1:KmlDtFcrdUzOYrBhXHgKw5ycWzc3ryPX5mQe0SkG3y8=
-k8s.io/client-go v0.29.0/go.mod h1:yLkXH4HKMAywcrD82KMSmfYg2DlE8mepPR4JGSo5n38=
+k8s.io/api v0.29.1 h1:DAjwWX/9YT7NQD4INu49ROJuZAAAP/Ijki48GUPzxqw=
+k8s.io/api v0.29.1/go.mod h1:7Kl10vBRUXhnQQI8YR/R327zXC8eJ7887/+Ybta+RoQ=
+k8s.io/apimachinery v0.29.1 h1:KY4/E6km/wLBguvCZv8cKTeOwwOBqFNjwJIdMkMbbRc=
+k8s.io/apimachinery v0.29.1/go.mod h1:6HVkd1FwxIagpYrHSwJlQqZI3G9LfYWRPAkUvLnXTKU=
+k8s.io/client-go v0.29.1 h1:19B/+2NGEwnFLzt0uB5kNJnfTsbV8w6TgQRz9l7ti7A=
+k8s.io/client-go v0.29.1/go.mod h1:TDG/psL9hdet0TI9mGyHJSgRkW3H9JZk2dNEUS7bRks=
 k8s.io/klog/v2 v2.110.1 h1:U/Af64HJf7FcwMcXyKm2RPM22WZzyR7OSpYj5tg3cL0=
 k8s.io/klog/v2 v2.110.1/go.mod h1:YGtd1984u+GgbuZ7e08/yBuAfKLSO0+uR1Fhi6ExXjo=
 k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 h1:aVUu9fTY98ivBPKR9Y5w/AuzbMm96cd3YHRTU83I780=
diff --git a/pb/events/v1/events.pb.go b/pb/events/v1/events.pb.go
index e62427b..156fa75 100644
--- a/pb/events/v1/events.pb.go
+++ b/pb/events/v1/events.pb.go
@@ -23,7 +23,7 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-// Deadletter defines the deadletter entry
+// Deadletter defines the deadletter event
 type Deadletter struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -167,7 +167,7 @@ func (x *ActorStarted) GetStartedAt() *timestamppb.Timestamp {
 	return nil
 }
 
-// ActorStopped defines the actor stopped events
+// ActorStopped defines the actor stopped event
 type ActorStopped struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -225,6 +225,7 @@ func (x *ActorStopped) GetStoppedAt() *timestamppb.Timestamp {
 	return nil
 }
 
+// ActorPassivated define the actor passivated event
 type ActorPassivated struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -282,6 +283,7 @@ func (x *ActorPassivated) GetPassivatedAt() *timestamppb.Timestamp {
 	return nil
 }
 
+// ActorChildCreated defines the child actor created event
 type ActorChildCreated struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -348,7 +350,7 @@ func (x *ActorChildCreated) GetCreatedAt() *timestamppb.Timestamp {
 	return nil
 }
 
-// ActorRestarted defines the actor restarted events
+// ActorRestarted defines the actor restarted event
 type ActorRestarted struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -406,6 +408,122 @@ func (x *ActorRestarted) GetRestartedAt() *timestamppb.Timestamp {
 	return nil
 }
 
+// NodeJoined defines the node joined event
+type NodeJoined struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Specifies the node address
+	Address string `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
+	// Specifies the timestamp
+	Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
+}
+
+func (x *NodeJoined) Reset() {
+	*x = NodeJoined{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_events_v1_events_proto_msgTypes[6]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *NodeJoined) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*NodeJoined) ProtoMessage() {}
+
+func (x *NodeJoined) ProtoReflect() protoreflect.Message {
+	mi := &file_events_v1_events_proto_msgTypes[6]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use NodeJoined.ProtoReflect.Descriptor instead.
+func (*NodeJoined) Descriptor() ([]byte, []int) {
+	return file_events_v1_events_proto_rawDescGZIP(), []int{6}
+}
+
+func (x *NodeJoined) GetAddress() string {
+	if x != nil {
+		return x.Address
+	}
+	return ""
+}
+
+func (x *NodeJoined) GetTimestamp() *timestamppb.Timestamp {
+	if x != nil {
+		return x.Timestamp
+	}
+	return nil
+}
+
+// NodeLeft defines the node left event
+type NodeLeft struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Specifies the node address
+	Address string `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
+	// Specifies the timestamp
+	Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
+}
+
+func (x *NodeLeft) Reset() {
+	*x = NodeLeft{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_events_v1_events_proto_msgTypes[7]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *NodeLeft) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*NodeLeft) ProtoMessage() {}
+
+func (x *NodeLeft) ProtoReflect() protoreflect.Message {
+	mi := &file_events_v1_events_proto_msgTypes[7]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use NodeLeft.ProtoReflect.Descriptor instead.
+func (*NodeLeft) Descriptor() ([]byte, []int) {
+	return file_events_v1_events_proto_rawDescGZIP(), []int{7}
+}
+
+func (x *NodeLeft) GetAddress() string {
+	if x != nil {
+		return x.Address
+	}
+	return ""
+}
+
+func (x *NodeLeft) GetTimestamp() *timestamppb.Timestamp {
+	if x != nil {
+		return x.Timestamp
+	}
+	return nil
+}
+
 var File_events_v1_events_proto protoreflect.FileDescriptor
 
 var file_events_v1_events_proto_rawDesc = []byte{
@@ -473,17 +591,29 @@ var file_events_v1_events_proto_rawDesc = []byte{
 	0x0c, 0x72, 0x65, 0x73, 0x74, 0x61, 0x72, 0x74, 0x65, 0x64, 0x5f, 0x61, 0x74, 0x18, 0x02, 0x20,
 	0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f,
 	0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52,
-	0x0b, 0x72, 0x65, 0x73, 0x74, 0x61, 0x72, 0x74, 0x65, 0x64, 0x41, 0x74, 0x42, 0x94, 0x01, 0x0a,
-	0x0d, 0x63, 0x6f, 0x6d, 0x2e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2e, 0x76, 0x31, 0x42, 0x0b,
-	0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x48, 0x02, 0x50, 0x01, 0x5a,
-	0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x74, 0x6f, 0x63, 0x68,
-	0x65, 0x6d, 0x65, 0x79, 0x2f, 0x67, 0x6f, 0x61, 0x6b, 0x74, 0x2f, 0x70, 0x62, 0x2f, 0x65, 0x76,
-	0x65, 0x6e, 0x74, 0x73, 0x2f, 0x76, 0x31, 0x3b, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x70, 0x62,
-	0xa2, 0x02, 0x03, 0x45, 0x58, 0x58, 0xaa, 0x02, 0x09, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2e,
-	0x56, 0x31, 0xca, 0x02, 0x09, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x5c, 0x56, 0x31, 0xe2, 0x02,
-	0x15, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x5c, 0x56, 0x31, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65,
-	0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x0a, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x3a,
-	0x3a, 0x56, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x0b, 0x72, 0x65, 0x73, 0x74, 0x61, 0x72, 0x74, 0x65, 0x64, 0x41, 0x74, 0x22, 0x60, 0x0a, 0x0a,
+	0x4e, 0x6f, 0x64, 0x65, 0x4a, 0x6f, 0x69, 0x6e, 0x65, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64,
+	0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64,
+	0x72, 0x65, 0x73, 0x73, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
+	0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74,
+	0x61, 0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x22, 0x5e,
+	0x0a, 0x08, 0x4e, 0x6f, 0x64, 0x65, 0x4c, 0x65, 0x66, 0x74, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64,
+	0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64,
+	0x72, 0x65, 0x73, 0x73, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
+	0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74,
+	0x61, 0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x42, 0x94,
+	0x01, 0x0a, 0x0d, 0x63, 0x6f, 0x6d, 0x2e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2e, 0x76, 0x31,
+	0x42, 0x0b, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x48, 0x02, 0x50,
+	0x01, 0x5a, 0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x74, 0x6f,
+	0x63, 0x68, 0x65, 0x6d, 0x65, 0x79, 0x2f, 0x67, 0x6f, 0x61, 0x6b, 0x74, 0x2f, 0x70, 0x62, 0x2f,
+	0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2f, 0x76, 0x31, 0x3b, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73,
+	0x70, 0x62, 0xa2, 0x02, 0x03, 0x45, 0x58, 0x58, 0xaa, 0x02, 0x09, 0x45, 0x76, 0x65, 0x6e, 0x74,
+	0x73, 0x2e, 0x56, 0x31, 0xca, 0x02, 0x09, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x5c, 0x56, 0x31,
+	0xe2, 0x02, 0x15, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x5c, 0x56, 0x31, 0x5c, 0x47, 0x50, 0x42,
+	0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x0a, 0x45, 0x76, 0x65, 0x6e, 0x74,
+	0x73, 0x3a, 0x3a, 0x56, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -498,7 +628,7 @@ func file_events_v1_events_proto_rawDescGZIP() []byte {
 	return file_events_v1_events_proto_rawDescData
 }
 
-var file_events_v1_events_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
+var file_events_v1_events_proto_msgTypes = make([]protoimpl.MessageInfo, 8)
 var file_events_v1_events_proto_goTypes = []interface{}{
 	(*Deadletter)(nil),            // 0: events.v1.Deadletter
 	(*ActorStarted)(nil),          // 1: events.v1.ActorStarted
@@ -506,31 +636,35 @@ var file_events_v1_events_proto_goTypes = []interface{}{
 	(*ActorPassivated)(nil),       // 3: events.v1.ActorPassivated
 	(*ActorChildCreated)(nil),     // 4: events.v1.ActorChildCreated
 	(*ActorRestarted)(nil),        // 5: events.v1.ActorRestarted
-	(*v1.Address)(nil),            // 6: address.v1.Address
-	(*anypb.Any)(nil),             // 7: google.protobuf.Any
-	(*timestamppb.Timestamp)(nil), // 8: google.protobuf.Timestamp
+	(*NodeJoined)(nil),            // 6: events.v1.NodeJoined
+	(*NodeLeft)(nil),              // 7: events.v1.NodeLeft
+	(*v1.Address)(nil),            // 8: address.v1.Address
+	(*anypb.Any)(nil),             // 9: google.protobuf.Any
+	(*timestamppb.Timestamp)(nil), // 10: google.protobuf.Timestamp
 }
 var file_events_v1_events_proto_depIdxs = []int32{
-	6,  // 0: events.v1.Deadletter.sender:type_name -> address.v1.Address
-	6,  // 1: events.v1.Deadletter.receiver:type_name -> address.v1.Address
-	7,  // 2: events.v1.Deadletter.message:type_name -> google.protobuf.Any
-	8,  // 3: events.v1.Deadletter.send_time:type_name -> google.protobuf.Timestamp
-	6,  // 4: events.v1.ActorStarted.address:type_name -> address.v1.Address
-	8,  // 5: events.v1.ActorStarted.started_at:type_name -> google.protobuf.Timestamp
-	6,  // 6: events.v1.ActorStopped.address:type_name -> address.v1.Address
-	8,  // 7: events.v1.ActorStopped.stopped_at:type_name -> google.protobuf.Timestamp
-	6,  // 8: events.v1.ActorPassivated.address:type_name -> address.v1.Address
-	8,  // 9: events.v1.ActorPassivated.passivated_at:type_name -> google.protobuf.Timestamp
-	6,  // 10: events.v1.ActorChildCreated.address:type_name -> address.v1.Address
-	6,  // 11: events.v1.ActorChildCreated.parent:type_name -> address.v1.Address
-	8,  // 12: events.v1.ActorChildCreated.created_at:type_name -> google.protobuf.Timestamp
-	6,  // 13: events.v1.ActorRestarted.address:type_name -> address.v1.Address
-	8,  // 14: events.v1.ActorRestarted.restarted_at:type_name -> google.protobuf.Timestamp
-	15, // [15:15] is the sub-list for method output_type
-	15, // [15:15] is the sub-list for method input_type
-	15, // [15:15] is the sub-list for extension type_name
-	15, // [15:15] is the sub-list for extension extendee
-	0,  // [0:15] is the sub-list for field type_name
+	8,  // 0: events.v1.Deadletter.sender:type_name -> address.v1.Address
+	8,  // 1: events.v1.Deadletter.receiver:type_name -> address.v1.Address
+	9,  // 2: events.v1.Deadletter.message:type_name -> google.protobuf.Any
+	10, // 3: events.v1.Deadletter.send_time:type_name -> google.protobuf.Timestamp
+	8,  // 4: events.v1.ActorStarted.address:type_name -> address.v1.Address
+	10, // 5: events.v1.ActorStarted.started_at:type_name -> google.protobuf.Timestamp
+	8,  // 6: events.v1.ActorStopped.address:type_name -> address.v1.Address
+	10, // 7: events.v1.ActorStopped.stopped_at:type_name -> google.protobuf.Timestamp
+	8,  // 8: events.v1.ActorPassivated.address:type_name -> address.v1.Address
+	10, // 9: events.v1.ActorPassivated.passivated_at:type_name -> google.protobuf.Timestamp
+	8,  // 10: events.v1.ActorChildCreated.address:type_name -> address.v1.Address
+	8,  // 11: events.v1.ActorChildCreated.parent:type_name -> address.v1.Address
+	10, // 12: events.v1.ActorChildCreated.created_at:type_name -> google.protobuf.Timestamp
+	8,  // 13: events.v1.ActorRestarted.address:type_name -> address.v1.Address
+	10, // 14: events.v1.ActorRestarted.restarted_at:type_name -> google.protobuf.Timestamp
+	10, // 15: events.v1.NodeJoined.timestamp:type_name -> google.protobuf.Timestamp
+	10, // 16: events.v1.NodeLeft.timestamp:type_name -> google.protobuf.Timestamp
+	17, // [17:17] is the sub-list for method output_type
+	17, // [17:17] is the sub-list for method input_type
+	17, // [17:17] is the sub-list for extension type_name
+	17, // [17:17] is the sub-list for extension extendee
+	0,  // [0:17] is the sub-list for field type_name
 }
 
 func init() { file_events_v1_events_proto_init() }
@@ -611,6 +745,30 @@ func file_events_v1_events_proto_init() {
 				return nil
 			}
 		}
+		file_events_v1_events_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*NodeJoined); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_events_v1_events_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*NodeLeft); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -618,7 +776,7 @@ func file_events_v1_events_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_events_v1_events_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   6,
+			NumMessages:   8,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
diff --git a/protos/goakt/events/v1/events.proto b/protos/goakt/events/v1/events.proto
index ffad2ed..85225b5 100644
--- a/protos/goakt/events/v1/events.proto
+++ b/protos/goakt/events/v1/events.proto
@@ -8,7 +8,7 @@ import "google/protobuf/timestamp.proto";
 
 option go_package = "github.com/tochemey/goakt/pb/events/v1;eventspb";
 
-// Deadletter defines the deadletter entry
+// Deadletter defines the deadletter event
 message Deadletter {
   // Specifies the sender's address
   address.v1.Address sender = 1;
@@ -31,7 +31,7 @@ message ActorStarted {
   google.protobuf.Timestamp started_at = 2;
 }
 
-// ActorStopped defines the actor stopped events
+// ActorStopped defines the actor stopped event
 message ActorStopped {
   // Specifies the actor address
   address.v1.Address address = 1;
@@ -39,6 +39,7 @@ message ActorStopped {
   google.protobuf.Timestamp stopped_at = 2;
 }
 
+// ActorPassivated define the actor passivated event
 message ActorPassivated {
   // Specifies the actor address
   address.v1.Address address = 1;
@@ -46,6 +47,7 @@ message ActorPassivated {
   google.protobuf.Timestamp passivated_at = 2;
 }
 
+// ActorChildCreated defines the child actor created event
 message ActorChildCreated {
   // Specifies the actor address
   address.v1.Address address = 1;
@@ -55,10 +57,26 @@ message ActorChildCreated {
   google.protobuf.Timestamp created_at = 3;
 }
 
-// ActorRestarted defines the actor restarted events
+// ActorRestarted defines the actor restarted event
 message ActorRestarted {
   // Specifies the actor address
   address.v1.Address address = 1;
   // Specifies the stop time
   google.protobuf.Timestamp restarted_at = 2;
 }
+
+// NodeJoined defines the node joined event
+message NodeJoined {
+  // Specifies the node address
+  string address = 1;
+  // Specifies the timestamp
+  google.protobuf.Timestamp timestamp = 2;
+}
+
+// NodeLeft defines the node left event
+message NodeLeft {
+  // Specifies the node address
+  string address = 1;
+  // Specifies the timestamp
+  google.protobuf.Timestamp timestamp = 2;
+}
\ No newline at end of file


Test patch:
diff --git a/actors/actor_system_test.go b/actors/actor_system_test.go
index c7e2423..1deca1d 100644
--- a/actors/actor_system_test.go
+++ b/actors/actor_system_test.go
@@ -405,6 +405,11 @@ func TestActorSystem(t *testing.T) {
 		err := sys.Start(ctx)
 		assert.NoError(t, err)
 
+		// create a deadletter subscriber
+		consumer, err := sys.Subscribe()
+		require.NoError(t, err)
+		require.NotNil(t, consumer)
+
 		actorName := "Exchanger"
 		actorRef, err := sys.Spawn(ctx, actorName, &Exchanger{})
 		assert.NoError(t, err)
@@ -429,6 +434,18 @@ func TestActorSystem(t *testing.T) {
 		time.Sleep(time.Second)
 		require.True(t, actorRef.IsRunning())
 
+		var items []*eventspb.ActorRestarted
+		for message := range consumer.Iterator() {
+			payload := message.Payload()
+			// only listening to deadletters
+			restarted, ok := payload.(*eventspb.ActorRestarted)
+			if ok {
+				items = append(items, restarted)
+			}
+		}
+
+		require.Len(t, items, 1)
+
 		t.Cleanup(func() {
 			err = sys.Stop(ctx)
 			assert.NoError(t, err)
@@ -879,8 +896,8 @@ func TestActorSystem(t *testing.T) {
 		require.NotNil(t, consumer)
 
 		// create the black hole actor
-		actor := &BlackHole{}
-		actorRef, err := sys.Spawn(ctx, "BlackHole ", actor)
+		actor := &Discarder{}
+		actorRef, err := sys.Spawn(ctx, "Discarder ", actor)
 		assert.NoError(t, err)
 		assert.NotNil(t, actorRef)
 
@@ -1091,4 +1108,84 @@ func TestActorSystem(t *testing.T) {
 			assert.NoError(t, err)
 		})
 	})
+	t.Run("With cluster events subscription", func(t *testing.T) {
+		// create a context
+		ctx := context.TODO()
+		// start the NATS server
+		srv := startNatsServer(t)
+
+		// create and start system cluster
+		cl1, sd1 := startClusterSystem(t, "Node1", srv.Addr().String())
+		peerAddress1 := cl1.PeerAddress()
+		require.NotEmpty(t, peerAddress1)
+
+		// create a subscriber to node 1
+		subscriber1, err := cl1.Subscribe()
+		require.NoError(t, err)
+		require.NotNil(t, subscriber1)
+
+		// create and start system cluster
+		cl2, sd2 := startClusterSystem(t, "Node2", srv.Addr().String())
+		peerAddress2 := cl2.PeerAddress()
+		require.NotEmpty(t, peerAddress2)
+
+		// create a subscriber to node 2
+		subscriber2, err := cl2.Subscribe()
+		require.NoError(t, err)
+		require.NotNil(t, subscriber2)
+
+		// wait for some time
+		time.Sleep(time.Second)
+
+		// capture the joins
+		var joins []*eventspb.NodeJoined
+		for event := range subscriber1.Iterator() {
+			// get the event payload
+			payload := event.Payload()
+			// only listening to cluster event
+			nodeJoined, ok := payload.(*eventspb.NodeJoined)
+			require.True(t, ok)
+			joins = append(joins, nodeJoined)
+		}
+
+		// assert the joins list
+		require.NotEmpty(t, joins)
+		require.Len(t, joins, 1)
+		require.Equal(t, peerAddress2, joins[0].GetAddress())
+
+		// wait for some time
+		time.Sleep(time.Second)
+
+		// stop the node
+		require.NoError(t, cl1.Unsubscribe(subscriber1))
+		assert.NoError(t, cl1.Stop(ctx))
+		assert.NoError(t, sd1.Close())
+
+		// wait for some time
+		time.Sleep(time.Second)
+
+		var lefts []*eventspb.NodeLeft
+		for event := range subscriber2.Iterator() {
+			payload := event.Payload()
+
+			// only listening to cluster event
+			nodeLeft, ok := payload.(*eventspb.NodeLeft)
+			require.True(t, ok)
+			lefts = append(lefts, nodeLeft)
+		}
+
+		require.NotEmpty(t, lefts)
+		require.Len(t, lefts, 1)
+		require.Equal(t, peerAddress1, lefts[0].GetAddress())
+
+		require.NoError(t, cl2.Unsubscribe(subscriber2))
+
+		t.Cleanup(func() {
+			assert.NoError(t, cl2.Stop(ctx))
+			// stop the discovery engines
+			assert.NoError(t, sd2.Close())
+			// shutdown the nats server gracefully
+			srv.Shutdown()
+		})
+	})
 }
diff --git a/actors/actor_test.go b/actors/actor_test.go
index c744782..f5d30c1 100644
--- a/actors/actor_test.go
+++ b/actors/actor_test.go
@@ -26,12 +26,20 @@ package actors
 
 import (
 	"context"
+	"os"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
 
+	natsserver "github.com/nats-io/nats-server/v2/server"
 	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+	"github.com/tochemey/goakt/discovery"
+	"github.com/tochemey/goakt/discovery/nats"
+	"github.com/tochemey/goakt/log"
 	testspb "github.com/tochemey/goakt/test/data/pb/v1"
+	"github.com/travisjeffery/go-dynaport"
 	"go.uber.org/atomic"
 	"go.uber.org/goleak"
 )
@@ -40,6 +48,8 @@ func TestMain(m *testing.M) {
 	goleak.VerifyTestMain(m, goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 		goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
 		goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+		goleak.IgnoreTopFunction("github.com/nats-io/nats%2ego.(*Conn).doReconnect"),
+		goleak.IgnoreTopFunction("sync.runtime_notifyListWait"),
 		goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"))
 }
 
@@ -349,21 +359,108 @@ func (x *Forwarder) PostStop(context.Context) error {
 
 var _ Actor = &Forwarder{}
 
-type BlackHole struct{}
+type Discarder struct{}
 
-func (d *BlackHole) PreStart(context.Context) error {
+var _ Actor = &Discarder{}
+
+func (d *Discarder) PreStart(context.Context) error {
 	return nil
 }
 
-func (d *BlackHole) Receive(ctx ReceiveContext) {
+func (d *Discarder) Receive(ctx ReceiveContext) {
 	switch ctx.Message().(type) {
 	default:
 		ctx.Unhandled()
 	}
 }
 
-func (d *BlackHole) PostStop(context.Context) error {
+func (d *Discarder) PostStop(context.Context) error {
 	return nil
 }
 
-var _ Actor = &BlackHole{}
+func startNatsServer(t *testing.T) *natsserver.Server {
+	t.Helper()
+	serv, err := natsserver.NewServer(&natsserver.Options{
+		Host: "127.0.0.1",
+		Port: -1,
+	})
+
+	require.NoError(t, err)
+
+	ready := make(chan bool)
+	go func() {
+		ready <- true
+		serv.Start()
+	}()
+	<-ready
+
+	if !serv.ReadyForConnections(2 * time.Second) {
+		t.Fatalf("nats-io server failed to start")
+	}
+
+	return serv
+}
+
+func startClusterSystem(t *testing.T, nodeName, serverAddr string) (ActorSystem, discovery.Provider) {
+	ctx := context.TODO()
+	logger := log.New(log.DebugLevel, os.Stdout)
+
+	// generate the ports for the single startNode
+	nodePorts := dynaport.Get(3)
+	gossipPort := nodePorts[0]
+	clusterPort := nodePorts[1]
+	remotingPort := nodePorts[2]
+
+	// create a Cluster startNode
+	host := "127.0.0.1"
+	// set the environments
+	require.NoError(t, os.Setenv("GOSSIP_PORT", strconv.Itoa(gossipPort)))
+	require.NoError(t, os.Setenv("CLUSTER_PORT", strconv.Itoa(clusterPort)))
+	require.NoError(t, os.Setenv("REMOTING_PORT", strconv.Itoa(remotingPort)))
+	require.NoError(t, os.Setenv("NODE_NAME", nodeName))
+	require.NoError(t, os.Setenv("NODE_IP", host))
+
+	// create the various config option
+	applicationName := "accounts"
+	actorSystemName := "testSystem"
+	natsSubject := "some-subject"
+	// create the instance of provider
+	provider := nats.NewDiscovery()
+
+	// create the config
+	config := discovery.Config{
+		nats.ApplicationName: applicationName,
+		nats.ActorSystemName: actorSystemName,
+		nats.NatsServer:      serverAddr,
+		nats.NatsSubject:     natsSubject,
+	}
+
+	// create the sd
+	sd := discovery.NewServiceDiscovery(provider, config)
+
+	// create the actor system
+	system, err := NewActorSystem(
+		nodeName,
+		WithPassivationDisabled(),
+		WithLogger(logger),
+		WithReplyTimeout(time.Minute),
+		WithClustering(sd, 10))
+
+	require.NotNil(t, system)
+	require.NoError(t, err)
+
+	// start the node
+	require.NoError(t, system.Start(ctx))
+
+	// clear the env var
+	require.NoError(t, os.Unsetenv("GOSSIP_PORT"))
+	require.NoError(t, os.Unsetenv("CLUSTER_PORT"))
+	require.NoError(t, os.Unsetenv("REMOTING_PORT"))
+	require.NoError(t, os.Unsetenv("NODE_NAME"))
+	require.NoError(t, os.Unsetenv("NODE_IP"))
+
+	time.Sleep(2 * time.Second)
+
+	// return the cluster startNode
+	return system, provider
+}
diff --git a/cluster/cluster_test.go b/cluster/node_test.go
similarity index 62%
rename from cluster/cluster_test.go
rename to cluster/node_test.go
index c50b312..0d8171b 100644
--- a/cluster/cluster_test.go
+++ b/cluster/node_test.go
@@ -27,26 +27,30 @@ package cluster
 import (
 	"context"
 	"fmt"
+	"os"
 	"strconv"
 	"testing"
 	"time"
 
 	"github.com/google/uuid"
+	natsserver "github.com/nats-io/nats-server/v2/server"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/tochemey/goakt/discovery"
+	"github.com/tochemey/goakt/discovery/nats"
 	internalpb "github.com/tochemey/goakt/internal/v1"
+	eventspb "github.com/tochemey/goakt/pb/events/v1"
 	testkit "github.com/tochemey/goakt/testkit/discovery"
 	"github.com/travisjeffery/go-dynaport"
 	"google.golang.org/protobuf/proto"
 )
 
-func TestCluster(t *testing.T) {
+func TestSingleNode(t *testing.T) {
 	t.Run("With Start and Shutdown", func(t *testing.T) {
 		// create the context
 		ctx := context.TODO()
 
-		// generate the ports for the single node
+		// generate the ports for the single startNode
 		nodePorts := dynaport.Get(3)
 		gossipPort := nodePorts[0]
 		clusterPort := nodePorts[1]
@@ -72,7 +76,7 @@ func TestCluster(t *testing.T) {
 		// create the service discovery
 		serviceDiscovery := discovery.NewServiceDiscovery(provider, config)
 
-		// create a Cluster node
+		// create a Node startNode
 		host := "localhost"
 
 		// set the environments
@@ -82,22 +86,22 @@ func TestCluster(t *testing.T) {
 		t.Setenv("NODE_NAME", "testNode")
 		t.Setenv("NODE_IP", host)
 
-		cluster, err := New("test", serviceDiscovery)
+		cluster, err := NewNode("test", serviceDiscovery)
 		require.NotNil(t, cluster)
 		require.NoError(t, err)
 
-		// start the Cluster node
+		// start the Node startNode
 		err = cluster.Start(ctx)
 		require.NoError(t, err)
 
 		hostNodeAddr := cluster.NodeHost()
 		assert.Equal(t, host, hostNodeAddr)
 
-		//  shutdown the Cluster node
+		//  shutdown the Node startNode
 		ctx, cancel := context.WithTimeout(ctx, time.Second)
 		defer cancel()
 
-		// stop the node
+		// stop the startNode
 		require.NoError(t, cluster.Stop(ctx))
 		provider.AssertExpectations(t)
 	})
@@ -105,7 +109,7 @@ func TestCluster(t *testing.T) {
 		// create the context
 		ctx := context.TODO()
 
-		// generate the ports for the single node
+		// generate the ports for the single startNode
 		nodePorts := dynaport.Get(3)
 		gossipPort := nodePorts[0]
 		clusterPort := nodePorts[1]
@@ -131,7 +135,7 @@ func TestCluster(t *testing.T) {
 		// create the service discovery
 		serviceDiscovery := discovery.NewServiceDiscovery(provider, config)
 
-		// create a Cluster node
+		// create a Node startNode
 		host := "localhost"
 		// set the environments
 		t.Setenv("GOSSIP_PORT", strconv.Itoa(gossipPort))
@@ -140,11 +144,11 @@ func TestCluster(t *testing.T) {
 		t.Setenv("NODE_NAME", "testNode")
 		t.Setenv("NODE_IP", host)
 
-		cluster, err := New("test", serviceDiscovery)
+		cluster, err := NewNode("test", serviceDiscovery)
 		require.NotNil(t, cluster)
 		require.NoError(t, err)
 
-		// start the Cluster node
+		// start the Node startNode
 		err = cluster.Start(ctx)
 		require.NoError(t, err)
 
@@ -152,7 +156,7 @@ func TestCluster(t *testing.T) {
 		actorName := uuid.NewString()
 		actor := &internalpb.WireActor{ActorName: actorName}
 
-		// replicate the actor in the Cluster
+		// replicate the actor in the Node
 		err = cluster.PutActor(ctx, actor)
 		require.NoError(t, err)
 
@@ -168,10 +172,10 @@ func TestCluster(t *testing.T) {
 		actual, err = cluster.GetActor(ctx, fakeActorName)
 		require.Nil(t, actual)
 		assert.EqualError(t, err, ErrActorNotFound.Error())
-		//  shutdown the Cluster node
+		//  shutdown the Node startNode
 		time.Sleep(time.Second)
 
-		// stop the node
+		// stop the startNode
 		require.NoError(t, cluster.Stop(ctx))
 		provider.AssertExpectations(t)
 	})
@@ -179,7 +183,7 @@ func TestCluster(t *testing.T) {
 		// create the context
 		ctx := context.TODO()
 
-		// generate the ports for the single node
+		// generate the ports for the single startNode
 		nodePorts := dynaport.Get(3)
 		gossipPort := nodePorts[0]
 		clusterPort := nodePorts[1]
@@ -205,7 +209,7 @@ func TestCluster(t *testing.T) {
 		// create the service discovery
 		serviceDiscovery := discovery.NewServiceDiscovery(provider, config)
 
-		// create a Cluster node
+		// create a Node startNode
 		host := "localhost"
 		// set the environments
 		t.Setenv("GOSSIP_PORT", strconv.Itoa(gossipPort))
@@ -214,11 +218,11 @@ func TestCluster(t *testing.T) {
 		t.Setenv("NODE_NAME", "testNode")
 		t.Setenv("NODE_IP", host)
 
-		cluster, err := New("test", serviceDiscovery)
+		cluster, err := NewNode("test", serviceDiscovery)
 		require.NotNil(t, cluster)
 		require.NoError(t, err)
 
-		// start the Cluster node
+		// start the Node startNode
 		err = cluster.Start(ctx)
 		require.NoError(t, err)
 
@@ -235,10 +239,10 @@ func TestCluster(t *testing.T) {
 		require.NoError(t, err)
 		assert.False(t, isSet)
 
-		//  shutdown the Cluster node
+		//  shutdown the Node startNode
 		time.Sleep(time.Second)
 
-		// stop the node
+		// stop the startNode
 		require.NoError(t, cluster.Stop(ctx))
 		provider.AssertExpectations(t)
 	})
@@ -246,7 +250,7 @@ func TestCluster(t *testing.T) {
 		// create the context
 		ctx := context.TODO()
 
-		// generate the ports for the single node
+		// generate the ports for the single startNode
 		nodePorts := dynaport.Get(3)
 		gossipPort := nodePorts[0]
 		clusterPort := nodePorts[1]
@@ -272,7 +276,7 @@ func TestCluster(t *testing.T) {
 		// create the service discovery
 		serviceDiscovery := discovery.NewServiceDiscovery(provider, config)
 
-		// create a Cluster node
+		// create a Node startNode
 		host := "localhost"
 		// set the environments
 		t.Setenv("GOSSIP_PORT", strconv.Itoa(gossipPort))
@@ -281,11 +285,11 @@ func TestCluster(t *testing.T) {
 		t.Setenv("NODE_NAME", "testNode")
 		t.Setenv("NODE_IP", host)
 
-		cluster, err := New("test", serviceDiscovery)
+		cluster, err := NewNode("test", serviceDiscovery)
 		require.NotNil(t, cluster)
 		require.NoError(t, err)
 
-		// start the Cluster node
+		// start the Node startNode
 		err = cluster.Start(ctx)
 		require.NoError(t, err)
 
@@ -293,7 +297,7 @@ func TestCluster(t *testing.T) {
 		actorName := uuid.NewString()
 		actor := &internalpb.WireActor{ActorName: actorName}
 
-		// replicate the actor in the Cluster
+		// replicate the actor in the Node
 		err = cluster.PutActor(ctx, actor)
 		require.NoError(t, err)
 
@@ -317,3 +321,166 @@ func TestCluster(t *testing.T) {
 		provider.AssertExpectations(t)
 	})
 }
+
+func TestMultipleNodes(t *testing.T) {
+	// start the NATS server
+	srv := startNatsServer(t)
+
+	// create a cluster node1
+	node1, sd1 := startNode(t, "node1", srv.Addr().String())
+	require.NotNil(t, node1)
+
+	// wait for the node to start properly
+	time.Sleep(2 * time.Second)
+
+	// create a cluster node1
+	node2, sd2 := startNode(t, "node2", srv.Addr().String())
+	require.NotNil(t, node2)
+	node2Addr := node2.AdvertisedAddress()
+
+	// wait for the node to start properly
+	time.Sleep(time.Second)
+
+	// assert the node joined cluster event
+	var events []*Event
+
+	// define an events reader loop and read events for some time
+L:
+	for {
+		select {
+		case event, ok := <-node1.Events():
+			if ok {
+				events = append(events, event)
+			}
+		case <-time.After(time.Second):
+			break L
+		}
+	}
+
+	require.NotEmpty(t, events)
+	require.Len(t, events, 1)
+	event := events[0]
+	msg, err := event.Type.UnmarshalNew()
+	require.NoError(t, err)
+	nodeJoined, ok := msg.(*eventspb.NodeJoined)
+	require.True(t, ok)
+	require.NotNil(t, nodeJoined)
+	require.Equal(t, node2Addr, nodeJoined.GetAddress())
+
+	// wait for some time
+	time.Sleep(time.Second)
+
+	// stop the second node
+	require.NoError(t, node2.Stop(context.TODO()))
+	// wait for the event to propagate properly
+	time.Sleep(time.Second)
+
+	// reset the slice
+	events = []*Event{}
+
+	// define an events reader loop and read events for some time
+L2:
+	for {
+		select {
+		case event, ok := <-node1.Events():
+			if ok {
+				events = append(events, event)
+			}
+		case <-time.After(time.Second):
+			break L2
+		}
+	}
+
+	require.NotEmpty(t, events)
+	require.Len(t, events, 1)
+	event = events[0]
+	msg, err = event.Type.UnmarshalNew()
+	require.NoError(t, err)
+	nodeLeft, ok := msg.(*eventspb.NodeLeft)
+	require.True(t, ok)
+	require.NotNil(t, nodeLeft)
+	require.Equal(t, node2Addr, nodeLeft.GetAddress())
+
+	t.Cleanup(func() {
+		require.NoError(t, node1.Stop(context.TODO()))
+		require.NoError(t, sd1.Close())
+		require.NoError(t, sd2.Close())
+		srv.Shutdown()
+	})
+}
+
+func startNatsServer(t *testing.T) *natsserver.Server {
+	t.Helper()
+	serv, err := natsserver.NewServer(&natsserver.Options{
+		Host: "127.0.0.1",
+		Port: -1,
+	})
+
+	require.NoError(t, err)
+
+	ready := make(chan bool)
+	go func() {
+		ready <- true
+		serv.Start()
+	}()
+	<-ready
+
+	if !serv.ReadyForConnections(2 * time.Second) {
+		t.Fatalf("nats-io server failed to start")
+	}
+
+	return serv
+}
+
+func startNode(t *testing.T, nodeName, serverAddr string) (*Node, discovery.Provider) {
+	// create a context
+	ctx := context.TODO()
+
+	// generate the ports for the single startNode
+	nodePorts := dynaport.Get(3)
+	gossipPort := nodePorts[0]
+	clusterPort := nodePorts[1]
+	remotingPort := nodePorts[2]
+
+	// create a Cluster startNode
+	host := "127.0.0.1"
+	// set the environments
+	require.NoError(t, os.Setenv("GOSSIP_PORT", strconv.Itoa(gossipPort)))
+	require.NoError(t, os.Setenv("CLUSTER_PORT", strconv.Itoa(clusterPort)))
+	require.NoError(t, os.Setenv("REMOTING_PORT", strconv.Itoa(remotingPort)))
+	require.NoError(t, os.Setenv("NODE_NAME", nodeName))
+	require.NoError(t, os.Setenv("NODE_IP", host))
+
+	// create the various config option
+	applicationName := "accounts"
+	actorSystemName := "testSystem"
+	natsSubject := "some-subject"
+	// create the instance of provider
+	provider := nats.NewDiscovery()
+
+	// create the config
+	config := discovery.Config{
+		nats.ApplicationName: applicationName,
+		nats.ActorSystemName: actorSystemName,
+		nats.NatsServer:      serverAddr,
+		nats.NatsSubject:     natsSubject,
+	}
+
+	// create the startNode
+	node, err := NewNode(nodeName, discovery.NewServiceDiscovery(provider, config))
+	require.NoError(t, err)
+	require.NotNil(t, node)
+
+	// start the node
+	require.NoError(t, node.Start(ctx))
+
+	// clear the env var
+	require.NoError(t, os.Unsetenv("GOSSIP_PORT"))
+	require.NoError(t, os.Unsetenv("CLUSTER_PORT"))
+	require.NoError(t, os.Unsetenv("REMOTING_PORT"))
+	require.NoError(t, os.Unsetenv("NODE_NAME"))
+	require.NoError(t, os.Unsetenv("NODE_IP"))
+
+	// return the cluster startNode
+	return node, provider
+}
diff --git a/cluster/option_test.go b/cluster/option_test.go
index d61e749..7302b60 100644
--- a/cluster/option_test.go
+++ b/cluster/option_test.go
@@ -38,43 +38,43 @@ func TestOptions(t *testing.T) {
 	testCases := []struct {
 		name     string
 		option   Option
-		expected Cluster
+		expected Node
 	}{
 		{
 			name:     "WithPartitionsCount",
 			option:   WithPartitionsCount(2),
-			expected: Cluster{partitionsCount: 2},
+			expected: Node{partitionsCount: 2},
 		},
 		{
 			name:     "WithLogger",
 			option:   WithLogger(log.DefaultLogger),
-			expected: Cluster{logger: log.DefaultLogger},
+			expected: Node{logger: log.DefaultLogger},
 		},
 		{
 			name:     "WithWriteTimeout",
 			option:   WithWriteTimeout(2 * time.Minute),
-			expected: Cluster{writeTimeout: 2 * time.Minute},
+			expected: Node{writeTimeout: 2 * time.Minute},
 		},
 		{
 			name:     "WithReadTimeout",
 			option:   WithReadTimeout(2 * time.Minute),
-			expected: Cluster{readTimeout: 2 * time.Minute},
+			expected: Node{readTimeout: 2 * time.Minute},
 		},
 		{
 			name:     "WithShutdownTimeout",
 			option:   WithShutdownTimeout(2 * time.Minute),
-			expected: Cluster{shutdownTimeout: 2 * time.Minute},
+			expected: Node{shutdownTimeout: 2 * time.Minute},
 		},
 		{
 			name:     "WithHasher",
 			option:   WithHasher(mockHasher),
-			expected: Cluster{hasher: mockHasher},
+			expected: Node{hasher: mockHasher},
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			var cl Cluster
+			var cl Node
 			tc.option.Apply(&cl)
 			assert.Equal(t, tc.expected, cl)
 		})
diff --git a/testkit/cluster/option.go b/testkit/cluster/option.go
index 72e540f..28529cc 100644
--- a/testkit/cluster/option.go
+++ b/testkit/cluster/option.go
@@ -21,7 +21,7 @@ func (_m *Option) EXPECT() *Option_Expecter {
 }
 
 // Apply provides a mock function with given fields: cl
-func (_m *Option) Apply(cl *goaktcluster.Cluster) {
+func (_m *Option) Apply(cl *goaktcluster.Node) {
 	_m.Called(cl)
 }
 
@@ -31,14 +31,14 @@ type Option_Apply_Call struct {
 }
 
 // Apply is a helper method to define mock.On call
-//   - cl *goaktcluster.Cluster
+//   - cl *goaktcluster.Node
 func (_e *Option_Expecter) Apply(cl interface{}) *Option_Apply_Call {
 	return &Option_Apply_Call{Call: _e.mock.On("Apply", cl)}
 }
 
-func (_c *Option_Apply_Call) Run(run func(cl *goaktcluster.Cluster)) *Option_Apply_Call {
+func (_c *Option_Apply_Call) Run(run func(cl *goaktcluster.Node)) *Option_Apply_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(*goaktcluster.Cluster))
+		run(args[0].(*goaktcluster.Node))
 	})
 	return _c
 }
@@ -48,7 +48,7 @@ func (_c *Option_Apply_Call) Return() *Option_Apply_Call {
 	return _c
 }
 
-func (_c *Option_Apply_Call) RunAndReturn(run func(*goaktcluster.Cluster)) *Option_Apply_Call {
+func (_c *Option_Apply_Call) RunAndReturn(run func(*goaktcluster.Node)) *Option_Apply_Call {
 	_c.Call.Return(run)
 	return _c
 }
diff --git a/testkit/cluster/option_func.go b/testkit/cluster/option_func.go
index b6992a9..b2e384e 100644
--- a/testkit/cluster/option_func.go
+++ b/testkit/cluster/option_func.go
@@ -21,7 +21,7 @@ func (_m *OptionFunc) EXPECT() *OptionFunc_Expecter {
 }
 
 // Execute provides a mock function with given fields: cl
-func (_m *OptionFunc) Execute(cl *goaktcluster.Cluster) {
+func (_m *OptionFunc) Execute(cl *goaktcluster.Node) {
 	_m.Called(cl)
 }
 
@@ -31,14 +31,14 @@ type OptionFunc_Execute_Call struct {
 }
 
 // Execute is a helper method to define mock.On call
-//   - cl *goaktcluster.Cluster
+//   - cl *goaktcluster.Node
 func (_e *OptionFunc_Expecter) Execute(cl interface{}) *OptionFunc_Execute_Call {
 	return &OptionFunc_Execute_Call{Call: _e.mock.On("Execute", cl)}
 }
 
-func (_c *OptionFunc_Execute_Call) Run(run func(cl *goaktcluster.Cluster)) *OptionFunc_Execute_Call {
+func (_c *OptionFunc_Execute_Call) Run(run func(cl *goaktcluster.Node)) *OptionFunc_Execute_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(*goaktcluster.Cluster))
+		run(args[0].(*goaktcluster.Node))
 	})
 	return _c
 }
@@ -48,7 +48,7 @@ func (_c *OptionFunc_Execute_Call) Return() *OptionFunc_Execute_Call {
 	return _c
 }
 
-func (_c *OptionFunc_Execute_Call) RunAndReturn(run func(*goaktcluster.Cluster)) *OptionFunc_Execute_Call {
+func (_c *OptionFunc_Execute_Call) RunAndReturn(run func(*goaktcluster.Node)) *OptionFunc_Execute_Call {
 	_c.Call.Return(run)
 	return _c
 }