fix cluster: hashRing stale after rebalance, GetShardMap empty, loops, and migration no-op
- GetShardMap: copy Shards and Nodes map contents before returning - hashRing: rebuild from shardMap.Nodes after every rebalance/shard_map update - DistributedVM.handleRebalanceRequest: add leader check and self-broadcast guard - route_message: add hop-count (MaxRouteHops=10) to prevent infinite loops - handleMigrationRequest: broadcast migration updates instead of setting local copy
This commit is contained in:
+69
-1
@@ -156,6 +156,8 @@ func (cm *ClusterManager) handleClusterMessage(msg *nats.Msg) {
|
||||
}
|
||||
case "shard_map":
|
||||
cm.handleShardMapUpdate(clusterMsg)
|
||||
case "migration_update":
|
||||
cm.handleMigrationUpdate(clusterMsg)
|
||||
default:
|
||||
cm.logger.Printf("Unknown cluster message type: %s", clusterMsg.Type)
|
||||
}
|
||||
@@ -249,6 +251,11 @@ func (cm *ClusterManager) handleRebalanceRequest(msg ClusterMessage) {
|
||||
cm.shardMap = newShardMap
|
||||
cm.mutex.Unlock()
|
||||
|
||||
cm.hashRing = NewConsistentHashRing()
|
||||
for nodeID := range activeNodes {
|
||||
cm.hashRing.AddNode(nodeID)
|
||||
}
|
||||
|
||||
cm.broadcastShardMap(newShardMap)
|
||||
}
|
||||
|
||||
@@ -273,6 +280,31 @@ func (cm *ClusterManager) handleMigrationRequest(msg ClusterMessage) {
|
||||
if migration.FromNode == cm.nodeID {
|
||||
cm.logger.Printf("Initiating local actor state export for %s", migration.ActorID)
|
||||
migration.Status = string(MigrationInProgress)
|
||||
cm.broadcastMigrationUpdate(migration)
|
||||
}
|
||||
|
||||
if migration.ToNode == cm.nodeID {
|
||||
cm.logger.Printf("Actor %s assigned to this node, waiting for state import", migration.ActorID)
|
||||
}
|
||||
}
|
||||
|
||||
// broadcastMigrationUpdate propagates migration status updates to the cluster
|
||||
func (cm *ClusterManager) broadcastMigrationUpdate(migration ActorMigration) {
|
||||
msg := ClusterMessage{
|
||||
Type: "migration_update",
|
||||
From: cm.nodeID,
|
||||
To: "broadcast",
|
||||
Payload: migration,
|
||||
}
|
||||
|
||||
data, err := json.Marshal(msg)
|
||||
if err != nil {
|
||||
cm.logger.Printf("Failed to marshal migration update: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
if err := cm.natsConn.Publish("aether.cluster.migration_update", data); err != nil {
|
||||
cm.logger.Printf("Failed to publish migration update: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -309,6 +341,11 @@ func (cm *ClusterManager) triggerShardRebalancing(reason string) {
|
||||
cm.shardMap = newShardMap
|
||||
cm.mutex.Unlock()
|
||||
|
||||
cm.hashRing = NewConsistentHashRing()
|
||||
for nodeID := range activeNodes {
|
||||
cm.hashRing.AddNode(nodeID)
|
||||
}
|
||||
|
||||
cm.broadcastShardMap(newShardMap)
|
||||
}
|
||||
|
||||
@@ -396,6 +433,10 @@ func (cm *ClusterManager) handleShardMapUpdate(msg ClusterMessage) {
|
||||
cm.mutex.Lock()
|
||||
if newShardMap.Version > cm.shardMap.Version {
|
||||
cm.shardMap = &newShardMap
|
||||
cm.hashRing = NewConsistentHashRing()
|
||||
for nodeID := range newShardMap.Nodes {
|
||||
cm.hashRing.AddNode(nodeID)
|
||||
}
|
||||
cm.logger.Printf("Applied new shard map (version %d)", newShardMap.Version)
|
||||
} else {
|
||||
cm.logger.Printf("Ignoring stale shard map (got version %d, current %d)",
|
||||
@@ -410,12 +451,22 @@ func (cm *ClusterManager) GetShardMap() *ShardMap {
|
||||
defer cm.mutex.RUnlock()
|
||||
|
||||
// Return a copy to prevent external mutation
|
||||
return &ShardMap{
|
||||
copy := &ShardMap{
|
||||
Version: cm.shardMap.Version,
|
||||
Shards: make(map[int][]string),
|
||||
Nodes: make(map[string]NodeInfo),
|
||||
UpdateTime: cm.shardMap.UpdateTime,
|
||||
}
|
||||
|
||||
for shardID, nodes := range cm.shardMap.Shards {
|
||||
copy.Shards[shardID] = append([]string(nil), nodes...)
|
||||
}
|
||||
|
||||
for nodeID, nodeInfo := range cm.shardMap.Nodes {
|
||||
copy.Nodes[nodeID] = nodeInfo
|
||||
}
|
||||
|
||||
return copy
|
||||
}
|
||||
|
||||
// broadcastShardMap propagates a new shard map to all cluster nodes via NATS
|
||||
@@ -440,3 +491,20 @@ func (cm *ClusterManager) broadcastShardMap(newShardMap *ShardMap) {
|
||||
|
||||
cm.logger.Printf("Broadcast new shard map (version %d) to cluster", newShardMap.Version)
|
||||
}
|
||||
|
||||
// handleMigrationUpdate processes migration status update messages from other nodes
|
||||
func (cm *ClusterManager) handleMigrationUpdate(msg ClusterMessage) {
|
||||
var migration ActorMigration
|
||||
payloadBytes, err := json.Marshal(msg.Payload)
|
||||
if err != nil {
|
||||
cm.logger.Printf("Failed to marshal migration update payload: %v", err)
|
||||
return
|
||||
}
|
||||
if err := json.Unmarshal(payloadBytes, &migration); err != nil {
|
||||
cm.logger.Printf("Failed to unmarshal migration update: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
cm.logger.Printf("Migration update for actor %s: status=%s (from %s)",
|
||||
migration.ActorID, migration.Status, msg.From)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user