Repository: Tochemey/goakt
Pull request #311 (2024-04-27 20:22:35)

refactor: minor code cleanup

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/311?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
Attention: Patch coverage is `56.16438%` with `32 lines` in your changes are missing coverage. Please review.
> Project coverage is 87.16%. Comparing base [(`670564f`)](https://app.codecov.io/gh/Tochemey/goakt/commit/670564fa2113c3f8a0f1f569a4ff61493ce06207?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) to head [(`215abb4`)](https://app.codecov.io/gh/Tochemey/goakt/pull/311?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).

| [Files](https://app.codecov.io/gh/Tochemey/goakt/pull/311?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) | Patch % | Lines |
|---|---|---|
| [actors/actor\_system.go](https://app.codecov.io/gh/Tochemey/goakt/pull/311?src=pr&el=tree&filepath=actors%2Factor_system.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL2FjdG9yX3N5c3RlbS5nbw==) | 27.50% | [28 Missing and 1 partial :warning: ](https://app.codecov.io/gh/Tochemey/goakt/pull/311?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) |
| [actors/registry.go](https://app.codecov.io/gh/Tochemey/goakt/pull/311?src=pr&el=tree&filepath=actors%2Fregistry.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL3JlZ2lzdHJ5Lmdv) | 90.90% | [3 Missing :warning: ](https://app.codecov.io/gh/Tochemey/goakt/pull/311?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     #311      +/-   ##
==========================================
- Coverage   87.90%   87.16%   -0.75%     
==========================================
  Files          31       31              
  Lines        3126     3186      +60     
==========================================
+ Hits         2748     2777      +29     
- Misses        266      296      +30     
- Partials      112      113       +1     
```



</details>

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


Patch:
diff --git a/actors/actor_system.go b/actors/actor_system.go
index 6a8b0c4..1bdb5c5 100644
--- a/actors/actor_system.go
+++ b/actors/actor_system.go
@@ -195,6 +195,7 @@ type actorSystem struct {
 	stashCapacity uint64
 
 	housekeeperStopSig chan types.Unit
+	clusterSyncStopSig chan types.Unit
 
 	// specifies the events stream
 	eventsStream *eventstream.EventsStream
@@ -246,6 +247,7 @@ func NewActorSystem(name string, opts ...Option) (ActorSystem, error) {
 		tracer:              noop.NewTracerProvider().Tracer(name),
 		eventsChan:          make(chan *cluster.Event, 1),
 		registry:            newRegistry(),
+		clusterSyncStopSig:  make(chan types.Unit, 1),
 	}
 
 	system.started.Store(false)
@@ -442,7 +444,7 @@ func (x *actorSystem) Spawn(ctx context.Context, name string, actor Actor) (PID,
 	}
 
 	x.actors.Set(pid)
-	x.registry.Register(actor)
+	x.registry.RegisterWithKey(name, actor)
 
 	if x.clusterEnabled.Load() {
 		actorType := reflect.TypeOf(actor).Elem().Name()
@@ -513,7 +515,7 @@ func (x *actorSystem) SpawnFromFunc(ctx context.Context, receiveFunc ReceiveFunc
 	}
 
 	x.actors.Set(pid)
-	x.registry.Register(actor)
+	x.registry.RegisterWithKey(actorID, actor)
 
 	if x.clusterEnabled.Load() {
 		actorType := reflect.TypeOf(actor).Elem().Name()
@@ -750,6 +752,9 @@ func (x *actorSystem) Start(ctx context.Context) error {
 		if err := x.enableClustering(spanCtx); err != nil {
 			return err
 		}
+		// start cluster synchronization
+		// TODO: revisit this
+		// go x.runClusterSync()
 	}
 
 	if x.remotingEnabled.Load() {
@@ -808,6 +813,7 @@ func (x *actorSystem) Stop(ctx context.Context) error {
 			return err
 		}
 		close(x.clusterChan)
+		x.clusterSyncStopSig <- types.Unit{}
 		x.clusterEnabled.Store(false)
 	}
 
@@ -1240,18 +1246,70 @@ func (x *actorSystem) registerMetrics() error {
 	return err
 }
 
-// broadcastClusterEvents listens to cluster events
+// broadcastClusterEvents listens to cluster events and send them to the event streams
 func (x *actorSystem) broadcastClusterEvents() {
 	for event := range x.eventsChan {
 		if x.clusterEnabled.Load() {
-			if event != nil && event.Type != nil {
-				message, _ := event.Type.UnmarshalNew()
+			if event != nil && event.Payload != nil {
+				// first need to resync actors map back to the cluster
+				x.clusterSync()
+				// push the event to the event stream
+				message, _ := event.Payload.UnmarshalNew()
 				if x.eventsStream != nil {
-					x.logger.Debugf("node=(%s) publishing cluster event=(%s)....", x.name, event.Type.GetTypeUrl())
+					x.logger.Debugf("node=(%s) publishing cluster event=(%s)....", x.name, event.Type)
 					x.eventsStream.Publish(eventsTopic, message)
-					x.logger.Debugf("cluster event=(%s) successfully published by node=(%s)", event.Type.GetTypeUrl(), x.name)
+					x.logger.Debugf("cluster event=(%s) successfully published by node=(%s)", event.Type, x.name)
 				}
 			}
 		}
 	}
 }
+
+// clusterSync synchronizes the node' actors map to the cluster.
+func (x *actorSystem) clusterSync() {
+	typesMap := x.registry.List()
+	if len(typesMap) != 0 {
+		x.logger.Info("syncing node actors map to the cluster...")
+		for actorID, actorType := range typesMap {
+			actorPath := NewPath(actorID, NewAddress(x.name, "", -1))
+			if x.remotingEnabled.Load() {
+				actorPath = NewPath(actorID, NewAddress(x.name, x.remotingHost, int(x.remotingPort)))
+			}
+
+			if !x.clusterEnabled.Load() {
+				return
+			}
+
+			x.clusterChan <- &internalpb.WireActor{
+				ActorName:    actorID,
+				ActorAddress: actorPath.RemoteAddress(),
+				ActorPath:    actorPath.String(),
+				ActorType:    actorType.Name(),
+			}
+		}
+		x.logger.Info("node actors map successfully synced back to the cluster.")
+	}
+}
+
+// runClusterSync runs time to time cluster synchronization
+// by populating the given node actors map to the cluster for availability
+func (x *actorSystem) runClusterSync() {
+	x.logger.Info("cluster synchronization has started...")
+	ticker := time.NewTicker(5 * time.Minute)
+	tickerStopSig := make(chan types.Unit, 1)
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				x.clusterSync()
+			case <-x.clusterSyncStopSig:
+				tickerStopSig <- types.Unit{}
+				return
+			}
+		}
+	}()
+
+	<-tickerStopSig
+	ticker.Stop()
+	x.logger.Info("cluster synchronization has stopped...")
+}
diff --git a/actors/registry.go b/actors/registry.go
index 778b4c0..abf23fd 100644
--- a/actors/registry.go
+++ b/actors/registry.go
@@ -33,23 +33,30 @@ import (
 // registry represents reflection registry for dynamic loading and creation of
 // actors at run-time
 type registry interface {
-	// Register an object with an alias/key
+	// Register an object
 	Register(v any)
+	// RegisterWithKey registers with an alias/key
+	RegisterWithKey(key string, v any)
 	// GetType returns the type of object,
 	GetType(v any) (reflect.Type, bool)
 	// GetTypeOf returns the type of an object name
 	GetTypeOf(name string) (reflect.Type, bool)
+	// GetTypeByKey returns the type of an object by key
+	GetTypeByKey(key string) (reflect.Type, bool)
 	// Deregister removes the registered object from the registry
 	Deregister(v any)
+	// DeregisterWithKey removes the registered object from the registry
+	DeregisterWithKey(key string)
 	// Exists return true when a given object is in the registry
 	Exists(v any) bool
+	// List returns the list of registered at any point in time
+	List() map[string]reflect.Type
 }
 
 // registryImpl implements Registry
 type registryImpl struct {
-	names  map[string]reflect.Type
-	rtypes map[string]reflect.Type
-	mu     sync.Mutex
+	typesMap map[string]reflect.Type
+	mu       sync.Mutex
 }
 
 // enforce compilation error
@@ -58,8 +65,8 @@ var _ registry = (*registryImpl)(nil)
 // newRegistry creates an instance of Registry
 func newRegistry() registry {
 	l := &registryImpl{
-		rtypes: make(map[string]reflect.Type),
-		mu:     sync.Mutex{},
+		typesMap: make(map[string]reflect.Type),
+		mu:       sync.Mutex{},
 	}
 	return l
 }
@@ -76,7 +83,7 @@ func (r *registryImpl) Register(v any) {
 
 	if _, exist := r.GetType(v); !exist {
 		r.mu.Lock()
-		r.rtypes[strings.ToLower(rtype.Name())] = rtype
+		r.typesMap[strings.ToLower(rtype.Name())] = rtype
 		r.mu.Unlock()
 	}
 }
@@ -85,14 +92,14 @@ func (r *registryImpl) Register(v any) {
 func (r *registryImpl) Deregister(v any) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	delete(r.rtypes, nameOf(v))
+	delete(r.typesMap, nameOf(v))
 }
 
 // GetType returns the type of object
 func (r *registryImpl) GetType(v any) (reflect.Type, bool) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	t, ok := r.rtypes[nameOf(v)]
+	t, ok := r.typesMap[nameOf(v)]
 	return t, ok
 }
 
@@ -100,7 +107,7 @@ func (r *registryImpl) GetType(v any) (reflect.Type, bool) {
 func (r *registryImpl) GetTypeOf(name string) (reflect.Type, bool) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	t, ok := r.rtypes[strings.ToLower(name)]
+	t, ok := r.typesMap[strings.ToLower(name)]
 	return t, ok
 }
 
@@ -108,7 +115,7 @@ func (r *registryImpl) GetTypeOf(name string) (reflect.Type, bool) {
 func (r *registryImpl) Exists(v any) bool {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	_, ok := r.rtypes[nameOf(v)]
+	_, ok := r.typesMap[nameOf(v)]
 	return ok
 }
 
@@ -123,3 +130,43 @@ func nameOf(v any) string {
 
 	return strings.ToLower(rtype.Name())
 }
+
+// List returns the list of registered at any point in time
+func (r *registryImpl) List() map[string]reflect.Type {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make(map[string]reflect.Type, len(r.typesMap))
+	for key, val := range r.typesMap {
+		out[key] = val
+	}
+	return out
+}
+
+// DeregisterWithKey implements registry.
+func (r *registryImpl) DeregisterWithKey(key string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	delete(r.typesMap, strings.ToLower(key))
+}
+
+// RegisterWithKey implements registry.
+func (r *registryImpl) RegisterWithKey(key string, v any) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var rtype reflect.Type
+	switch _type := v.(type) {
+	case reflect.Type:
+		rtype = _type
+	default:
+		rtype = reflect.TypeOf(v).Elem()
+	}
+	r.typesMap[strings.ToLower(key)] = rtype
+}
+
+// GetTypeByKey returns the type of an object by key
+func (r *registryImpl) GetTypeByKey(key string) (reflect.Type, bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	t, ok := r.typesMap[strings.ToLower(key)]
+	return t, ok
+}
diff --git a/internal/cluster/node.go b/internal/cluster/node.go
index 0910df3..4326ac6 100644
--- a/internal/cluster/node.go
+++ b/internal/cluster/node.go
@@ -27,11 +27,11 @@ package cluster
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"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/pkg/errors"
@@ -47,9 +47,28 @@ import (
 	"github.com/tochemey/goakt/log"
 )
 
+type EventType int
+
+const (
+	NodeJoined EventType = iota
+	NodeLeft
+)
+
+func (x EventType) String() string {
+	switch x {
+	case NodeJoined:
+		return "NodJoined"
+	case NodeLeft:
+		return "NodeLeft"
+	default:
+		return fmt.Sprintf("%d", int(x))
+	}
+}
+
 // Event defines the cluster event
 type Event struct {
-	Type *anypb.Any
+	Payload *anypb.Any
+	Type    EventType
 }
 
 // Interface defines the Node interface
@@ -173,7 +192,7 @@ func (n *Node) Start(ctx context.Context) error {
 	conf := n.buildConfig()
 	conf.Hasher = &hasherWrapper{n.hasher}
 
-	m, err := olriconfig.NewMemberlistConfig("lan")
+	m, err := config.NewMemberlistConfig("lan")
 	if err != nil {
 		logger.Error(errors.Wrap(err, "failed to configure the cluster Node memberlist.💥"))
 		return err
@@ -202,7 +221,6 @@ func (n *Node) Start(ctx context.Context) error {
 	// cancel the context the server has started
 	conf.Started = func() {
 		defer cancel()
-		logger.Infof("GoAkt cluster Node=(%s) successfully started. 🎉", n.name)
 	}
 
 	eng, err := olric.New(conf)
@@ -224,6 +242,7 @@ func (n *Node) Start(ctx context.Context) error {
 	}()
 
 	<-startCtx.Done()
+	logger.Info("cluster engine successfully started. 🎉")
 
 	// set the client
 	n.client = n.server.NewEmbeddedClient()
@@ -246,6 +265,7 @@ func (n *Node) Start(ctx context.Context) error {
 	n.messagesChan = n.pubSub.Channel()
 	go n.consume()
 
+	logger.Infof("GoAkt cluster Node=(%s) successfully started. 🎉", n.name)
 	return nil
 }
 
@@ -479,8 +499,8 @@ func (n *Node) consume() {
 
 				n.logger.Debugf("%s received (%s) cluster event:[addr=(%s)]", n.name, kind, event.GetAddress())
 
-				eventType, _ := anypb.New(event)
-				n.events <- &Event{eventType}
+				payload, _ := anypb.New(event)
+				n.events <- &Event{payload, NodeJoined}
 
 			case events.KindNodeLeftEvent:
 				nodeLeft := new(events.NodeLeftEvent)
@@ -500,8 +520,8 @@ func (n *Node) consume() {
 
 				n.logger.Debugf("%s received (%s) cluster event:[addr=(%s)]", n.name, kind, event.GetAddress())
 
-				eventType, _ := anypb.New(event)
-				n.events <- &Event{eventType}
+				payload, _ := anypb.New(event)
+				n.events <- &Event{payload, NodeLeft}
 			default:
 				// skip
 			}
@@ -534,11 +554,11 @@ func (n *Node) buildConfig() *config.Config {
 		ReadQuorum:                 config.DefaultReadQuorum,
 		MemberCountQuorum:          config.DefaultMemberCountQuorum,
 		Peers:                      []string{},
-		DMaps:                      &olriconfig.DMaps{},
+		DMaps:                      &config.DMaps{},
 		KeepAlivePeriod:            config.DefaultKeepAlivePeriod,
 		PartitionCount:             n.partitionsCount,
 		BootstrapTimeout:           config.DefaultBootstrapTimeout,
-		ReplicationMode:            olriconfig.SyncReplicationMode,
+		ReplicationMode:            config.SyncReplicationMode,
 		RoutingTablePushInterval:   config.DefaultRoutingTablePushInterval,
 		JoinRetryInterval:          config.DefaultJoinRetryInterval,
 		MaxJoinAttempts:            config.DefaultMaxJoinAttempts,


Test patch:
diff --git a/actors/registry_test.go b/actors/registry_test.go
index 8713299..25edffc 100644
--- a/actors/registry_test.go
+++ b/actors/registry_test.go
@@ -39,15 +39,43 @@ func TestRegistry(t *testing.T) {
 	})
 
 	t.Run("With registration", func(t *testing.T) {
-		newRegistry := newRegistry()
+		registry := newRegistry()
 		// create an instance of an object
 		actor := newTestActor()
 		// register that actor
-		newRegistry.Register(actor)
-		_, ok := newRegistry.GetType(actor)
+		registry.Register(actor)
+		_, ok := registry.GetType(actor)
 		assert.True(t, ok)
 
-		_, ok = newRegistry.GetTypeOf("testActor")
+		_, ok = registry.GetTypeOf("testActor")
 		assert.True(t, ok)
+		assert.Len(t, registry.List(), 1)
+
+		registry.Deregister(actor)
+		assert.Len(t, registry.List(), 0)
+	})
+
+	t.Run("With registration/deregistration with key", func(t *testing.T) {
+		registry := newRegistry()
+
+		actor := newTestActor()
+		key := "key-1"
+		registry.RegisterWithKey(key, actor)
+		registry.Register(actor)
+
+		_, ok := registry.GetType(actor)
+		assert.True(t, ok)
+		_, ok = registry.GetTypeByKey(key)
+		assert.True(t, ok)
+
+		_, ok = registry.GetTypeOf("testActor")
+		assert.True(t, ok)
+		assert.Len(t, registry.List(), 2)
+
+		registry.Deregister(actor)
+		assert.Len(t, registry.List(), 1)
+
+		registry.DeregisterWithKey(key)
+		assert.Empty(t, registry.List())
 	})
 }
diff --git a/internal/cluster/node_test.go b/internal/cluster/node_test.go
index a341b38..5310c02 100644
--- a/internal/cluster/node_test.go
+++ b/internal/cluster/node_test.go
@@ -398,7 +398,7 @@ L:
 	require.NotEmpty(t, events)
 	require.Len(t, events, 1)
 	event := events[0]
-	msg, err := event.Type.UnmarshalNew()
+	msg, err := event.Payload.UnmarshalNew()
 	require.NoError(t, err)
 	nodeJoined, ok := msg.(*goaktpb.NodeJoined)
 	require.True(t, ok)
@@ -432,7 +432,7 @@ L2:
 	require.NotEmpty(t, events)
 	require.Len(t, events, 1)
 	event = events[0]
-	msg, err = event.Type.UnmarshalNew()
+	msg, err = event.Payload.UnmarshalNew()
 	require.NoError(t, err)
 	nodeLeft, ok := msg.(*goaktpb.NodeLeft)
 	require.True(t, ok)