diff --git a/cluster/distributed.go b/cluster/distributed.go index 4d779ab..68e7d14 100644 --- a/cluster/distributed.go +++ b/cluster/distributed.go @@ -151,8 +151,10 @@ func (dvm *DistributedVM) SendMessage(message RuntimeMessage) error { // routeMessageToNode sends a message to another node for delivery to the target actor func (dvm *DistributedVM) routeMessageToNode(actorID string, message RuntimeMessage) error { hops := 0 + var body map[string]interface{} if mp, ok := message.(*MessagePayload); ok { hops = mp.Hops + body = mp.Body } if hops >= MaxRouteHops { dvm.cluster.logger.Printf("Dropping message for actor %s: exceeded max hops (%d)", actorID, MaxRouteHops) @@ -167,6 +169,7 @@ func (dvm *DistributedVM) routeMessageToNode(actorID string, message RuntimeMess TargetActorID: actorID, Type: message.GetType(), Hops: hops + 1, + Body: body, }, Timestamp: time.Now(), } @@ -219,8 +222,10 @@ func (dvm *DistributedVM) handleClusterMessage(msg *nats.Msg) { dvm.localRuntime.LoadModel(&model) case "route_message": - // Handle message routing from other nodes - // Re-marshal and unmarshal to convert map[string]interface{} to concrete type + if clusterMsg.From == dvm.nodeID { + return + } + payloadBytes, err := json.Marshal(clusterMsg.Payload) if err != nil { return @@ -236,11 +241,16 @@ func (dvm *DistributedVM) handleClusterMessage(msg *nats.Msg) { } targetActor := message.TargetActorID + msg := &MessagePayload{ + TargetActorID: targetActor, + Type: message.Type, + Hops: message.Hops, + Body: message.Body, + } if dvm.IsLocalActor(targetActor) { - dvm.localRuntime.SendMessage(&message) + dvm.localRuntime.SendMessage(msg) } else { - // Relay to the correct node - dvm.routeMessageToNode(targetActor, &message) + dvm.routeMessageToNode(targetActor, msg) } case "rebalance": @@ -260,6 +270,11 @@ func (dvm *DistributedVM) handleRebalanceRequest(msg ClusterMessage) { return } + if dvm.cluster.shardMap == nil { + dvm.cluster.logger.Printf("Shard map is nil, skipping rebalance") + return + } + payloadBytes, err := json.Marshal(msg.Payload) if err != nil { dvm.cluster.logger.Printf("Failed to marshal rebalance payload: %v", err) diff --git a/cluster/shard.go b/cluster/shard.go index 64cf425..17a1d29 100644 --- a/cluster/shard.go +++ b/cluster/shard.go @@ -195,10 +195,12 @@ func (chp *ConsistentHashPlacement) PlaceActor(actorID string, shardMap *ShardMa node := ring.GetNode(actorID) if node == "" { + sortedNodeIDs := make([]string, 0, len(nodes)) for nodeID := range nodes { - return nodeID, nil + sortedNodeIDs = append(sortedNodeIDs, nodeID) } - return "", fmt.Errorf("failed to place actor") + sort.Strings(sortedNodeIDs) + return sortedNodeIDs[0], nil } return node, nil @@ -231,10 +233,12 @@ func (chp *ConsistentHashPlacement) RebalanceShards(currentMap *ShardMap, nodes for shardID := range currentMap.Shards { primaryNode := ring.GetNode(fmt.Sprintf("shard-%d", shardID)) if primaryNode == "" { + sortedNodeIDs := make([]string, 0, len(nodes)) for nodeID := range nodes { - primaryNode = nodeID - break + sortedNodeIDs = append(sortedNodeIDs, nodeID) } + sort.Strings(sortedNodeIDs) + primaryNode = sortedNodeIDs[0] } var replicaNodes []string diff --git a/cluster/types.go b/cluster/types.go index ee7630e..a10af78 100644 --- a/cluster/types.go +++ b/cluster/types.go @@ -197,9 +197,10 @@ const MaxRouteHops = 10 // MessagePayload is a concrete type for JSON-unmarshaling RuntimeMessage payloads. // Use this when receiving message data over the network. type MessagePayload struct { - TargetActorID string `json:"targetActorId"` - Type string `json:"type"` - Hops int `json:"hops,omitempty"` + TargetActorID string `json:"targetActorId"` + Type string `json:"type"` + Hops int `json:"hops,omitempty"` + Body map[string]interface{} `json:"body,omitempty"` } // GetTargetActorID implements RuntimeMessage