fix: fixes

Author: Tochemey

actor/cluster_config.go

@@ -451,7 +451,8 @@ func (x *ClusterConfig) WithDataCenter(config *datacenter.Config) *ClusterConfig
 
 // WithCRDT enables CRDT replication on this cluster node.
 // When set, the actor system spawns a Replicator system actor that subscribes
-// to CRDT key topics via the TopicActor and replicates state across the cluster.
+// to the shared goakt.crdt.deltas topic via the TopicActor and replicates
+// state across the cluster.
 // If not called, no Replicator is spawned and there is zero CRDT overhead.
 func (x *ClusterConfig) WithCRDT(opts ...crdt.Option) *ClusterConfig {
 	x.crdtConfig = crdt.NewConfig(opts...)

actor/replicator.go

@@ -96,10 +96,10 @@ func (e *crdtConfigExtension) Config() *crdt.Config {
 // and replicates state across the cluster via TopicActor pub/sub.
 //
 // Each node in the cluster runs its own replicatorActor. All replicators
-// subscribe to the same CRDT key topics via the TopicActor. When any
-// replicator updates a key, it publishes the delta to that key's topic.
-// Because every replicator is subscribed to the same topic, they all
-// receive the delta automatically.
+// subscribe to a single shared topic (goakt.crdt.deltas) via the TopicActor.
+// When any replicator updates a key, it publishes the delta (which carries
+// the key inside the payload) to this shared topic. Because every replicator
+// is subscribed to the same topic, they all receive the delta automatically.
 type replicatorActor struct {
 	pid                   *PID
 	topicActor            *PID

crdt/config.go

@@ -127,7 +127,7 @@ func WithTombstoneTTL(duration time.Duration) Option {
 
 // WithRole restricts CRDT replication to cluster nodes that advertise
 // the specified role. Only nodes with this role will spawn a Replicator
-// and subscribe to CRDT key topics. An empty string means all nodes participate.
+// and subscribe to the shared CRDT delta topic. An empty string means all nodes participate.
 func WithRole(role string) Option {
 	return func(c *Config) {
 		c.role = role

crdt/key.go

@@ -46,7 +46,8 @@ const (
 //
 // The generic parameter T binds the key to a specific CRDT type at compile time,
 // providing type safety for Update, Get, Subscribe, and Changed messages.
-// The key's ID is used as the TopicActor topic name for replication.
+// The key's ID is carried inside each delta payload for routing; all deltas
+// are published to a single shared topic (goakt.crdt.deltas) via TopicActor.
 // The DataType is serialized in anti-entropy and coordination messages
 // so that peers can validate type consistency.
 //
@@ -60,7 +61,8 @@ type Key[T ReplicatedData] struct {
 }
 
 // ID returns the key's string identifier.
-// This is used as the TopicActor topic name and the internal store key.
+// This is used as the internal store key and is embedded in delta messages
+// for routing; replication uses the shared goakt.crdt.deltas topic.
 func (k Key[T]) ID() string {
 	return k.id
 }

crdt/messages.go

@@ -25,7 +25,8 @@ package crdt
 // Update is sent to the Replicator to create or update a CRDT key.
 //
 // The update is always applied locally first and the delta is published
-// to the key's topic via TopicActor. If WriteTo is set, the Replicator
+// to the shared goakt.crdt.deltas topic via TopicActor (the key is
+// carried inside the delta payload). If WriteTo is set, the Replicator
 // also sends the delta directly to peers and waits for acknowledgments
 // before returning the response.
 //
@@ -79,8 +80,8 @@ type Changed[T ReplicatedData] struct {
 }
 
 // Delete is sent to the Replicator to remove a CRDT key.
-// Deletion publishes a tombstone to the key's topic. Tombstones are
-// retained for the configured TombstoneTTL before pruning.
+// Deletion publishes a tombstone to the shared goakt.crdt.deltas topic.
+// Tombstones are retained for the configured TombstoneTTL before pruning.
 type Delete[T ReplicatedData] struct {
 	Key     Key[T]
 	WriteTo Coordination

docs/advanced/distributed-data.mdx

@@ -106,7 +106,7 @@ func (a *RateLimiter) Receive(ctx *actor.ReceiveContext) {
             Key:     crdt.PNCounterKey("request-count"),
             Initial: crdt.NewPNCounter(),
             Modify: func(current *crdt.PNCounter) *crdt.PNCounter {
-                return current.Increment(ctx.Self().Path().Name(), 1)
+                return current.Increment(ctx.ActorSystem().PeersAddress(), 1)
             },
         })
     }
@@ -283,7 +283,7 @@ func (a *SessionTracker) Receive(ctx *actor.ReceiveContext) {
 
     switch msg := ctx.Message().(type) {
     case *AddSession:
-        nodeID := ctx.Self().Path().Name()
+        nodeID := ctx.ActorSystem().PeersAddress()
         ctx.Tell(replicator, &crdt.Update[*crdt.ORSet[string]]{
             Key:     sessionsKey,
             Initial: crdt.NewORSet[string](),

internal/internalpb/crdt.pb.go

@@ -127,8 +127,9 @@ message MVRegisterData {
   map<string, uint64> clock = 2;
 }
 
-// CRDTDelta is the delta message published to a key's topic via TopicActor.
-// All Replicators subscribed to the same key topic receive this message.
+// CRDTDelta is the delta message published to the shared goakt.crdt.deltas
+// topic via TopicActor. The key is carried inside the payload so that
+// receivers can route the delta to the correct local store entry.
 message CRDTDelta {
   // Specifies the CRDT key this delta belongs to.
   CRDTKey key = 1;