implement cluster stubs: cross-node routing, rebalancing, and actor migration

- Fix ConsistentHashPlacement.PlaceActor() to use consistent hash ring
- Implement ConsistentHashPlacement.RebalanceShards() to redistribute shards
- Implement ClusterManager.handleRebalanceRequest() with actual rebalancing
- Implement ClusterManager.handleMigrationRequest() for actor state transfer
- Implement ClusterManager.triggerShardRebalancing() to compute and broadcast
- Implement DistributedVM.SendMessage() with cross-node NATS routing
- Implement DistributedVM.handleRebalanceRequest() to update shard map
- Fix route_message handler to check if actor is local before delivery
- Update ConsistentHashPlacement.RebalanceShards() test for new behavior
- Add handleShardMapUpdate() and broadcastShardMap() to ClusterManager

cluster/manager.go

@@ -154,6 +154,8 @@ func (cm *ClusterManager) handleClusterMessage(msg *nats.Msg) {
 		if update, ok := clusterMsg.Payload.(NodeUpdate); ok {
 			cm.handleNodeUpdate(update)
 		}
+	case "shard_map":
+		cm.handleShardMapUpdate(clusterMsg)
 	default:
 		cm.logger.Printf("Unknown cluster message type: %s", clusterMsg.Type)
 	}
@@ -217,16 +219,61 @@ func (cm *ClusterManager) handleNodeUpdate(update NodeUpdate) {
 func (cm *ClusterManager) handleRebalanceRequest(msg ClusterMessage) {
 	cm.logger.Printf("Handling rebalance request from %s", msg.From)
 
-	// Implementation would handle the specific rebalancing logic
-	// This is a simplified version
+	if !cm.IsLeader() {
+		cm.logger.Printf("Ignoring rebalance request: not the leader")
+		return
+	}
+
+	cm.mutex.RLock()
+	activeNodes := make(map[string]*NodeInfo)
+	for nodeID, nodeInfo := range cm.nodes {
+		if nodeInfo.Status == NodeStatusActive {
+			activeNodes[nodeID] = nodeInfo
+		}
+	}
+	cm.mutex.RUnlock()
+
+	if len(activeNodes) == 0 {
+		cm.logger.Printf("No active nodes for rebalancing")
+		return
+	}
+
+	placement := &ConsistentHashPlacement{}
+	newShardMap, err := placement.RebalanceShards(cm.shardMap, activeNodes)
+	if err != nil {
+		cm.logger.Printf("Failed to compute new shard map: %v", err)
+		return
+	}
+
+	cm.mutex.Lock()
+	cm.shardMap = newShardMap
+	cm.mutex.Unlock()
+
+	cm.broadcastShardMap(newShardMap)
 }
 
 // handleMigrationRequest processes actor migration requests
 func (cm *ClusterManager) handleMigrationRequest(msg ClusterMessage) {
 	cm.logger.Printf("Handling migration request from %s", msg.From)
 
-	// Implementation would handle the specific migration logic
-	// This is a simplified version
+	var migration ActorMigration
+	payloadBytes, err := json.Marshal(msg.Payload)
+	if err != nil {
+		cm.logger.Printf("Failed to marshal migration payload: %v", err)
+		return
+	}
+	if err := json.Unmarshal(payloadBytes, &migration); err != nil {
+		cm.logger.Printf("Failed to unmarshal migration request: %v", err)
+		return
+	}
+
+	cm.logger.Printf("Actor %s migrating from %s to %s (shard %d)",
+		migration.ActorID, migration.FromNode, migration.ToNode, migration.ShardID)
+
+	if migration.FromNode == cm.nodeID {
+		cm.logger.Printf("Initiating local actor state export for %s", migration.ActorID)
+		migration.Status = string(MigrationInProgress)
+	}
 }
 
 // triggerShardRebalancing initiates shard rebalancing across the cluster
@@ -237,12 +284,11 @@ func (cm *ClusterManager) triggerShardRebalancing(reason string) {
 
 	cm.logger.Printf("Triggering shard rebalancing: %s", reason)
 
-	// Get active nodes
-	var activeNodes []*NodeInfo
 	cm.mutex.RLock()
-	for _, node := range cm.nodes {
-		if node.Status == NodeStatusActive {
-			activeNodes = append(activeNodes, node)
+	activeNodes := make(map[string]*NodeInfo)
+	for nodeID, nodeInfo := range cm.nodes {
+		if nodeInfo.Status == NodeStatusActive {
+			activeNodes[nodeID] = nodeInfo
 		}
 	}
 	cm.mutex.RUnlock()
@@ -252,8 +298,18 @@ func (cm *ClusterManager) triggerShardRebalancing(reason string) {
 		return
 	}
 
-	// This would implement the actual rebalancing logic
-	cm.logger.Printf("Would rebalance across %d active nodes", len(activeNodes))
+	placement := &ConsistentHashPlacement{}
+	newShardMap, err := placement.RebalanceShards(cm.shardMap, activeNodes)
+	if err != nil {
+		cm.logger.Printf("Failed to compute new shard map: %v", err)
+		return
+	}
+
+	cm.mutex.Lock()
+	cm.shardMap = newShardMap
+	cm.mutex.Unlock()
+
+	cm.broadcastShardMap(newShardMap)
 }
 
 // monitorNodes periodically checks node health and updates
@@ -319,6 +375,35 @@ func (cm *ClusterManager) GetNodes() map[string]*NodeInfo {
 	return nodes
 }
 
+// handleShardMapUpdate applies a new shard map received from the leader
+func (cm *ClusterManager) handleShardMapUpdate(msg ClusterMessage) {
+	if msg.From == cm.nodeID {
+		return
+	}
+
+	payloadBytes, err := json.Marshal(msg.Payload)
+	if err != nil {
+		cm.logger.Printf("Failed to marshal shard map payload: %v", err)
+		return
+	}
+
+	var newShardMap ShardMap
+	if err := json.Unmarshal(payloadBytes, &newShardMap); err != nil {
+		cm.logger.Printf("Failed to unmarshal shard map: %v", err)
+		return
+	}
+
+	cm.mutex.Lock()
+	if newShardMap.Version > cm.shardMap.Version {
+		cm.shardMap = &newShardMap
+		cm.logger.Printf("Applied new shard map (version %d)", newShardMap.Version)
+	} else {
+		cm.logger.Printf("Ignoring stale shard map (got version %d, current %d)",
+			newShardMap.Version, cm.shardMap.Version)
+	}
+	cm.mutex.Unlock()
+}
+
 // GetShardMap returns the current shard mapping
 func (cm *ClusterManager) GetShardMap() *ShardMap {
 	cm.mutex.RLock()
@@ -332,3 +417,26 @@ func (cm *ClusterManager) GetShardMap() *ShardMap {
 		UpdateTime: cm.shardMap.UpdateTime,
 	}
 }
+
+// broadcastShardMap propagates a new shard map to all cluster nodes via NATS
+func (cm *ClusterManager) broadcastShardMap(newShardMap *ShardMap) {
+	msg := ClusterMessage{
+		Type:      "shard_map",
+		From:      cm.nodeID,
+		To:        "broadcast",
+		Payload:   newShardMap,
+		Timestamp: time.Now(),
+	}
+
+	data, err := json.Marshal(msg)
+	if err != nil {
+		cm.logger.Printf("Failed to marshal shard map broadcast: %v", err)
+		return
+	}
+
+	if err := cm.natsConn.Publish("aether.cluster.shard_map", data); err != nil {
+		cm.logger.Printf("Failed to publish shard map broadcast: %v", err)
+	}
+
+	cm.logger.Printf("Broadcast new shard map (version %d) to cluster", newShardMap.Version)
+}