Module: Legion::Extensions::Mesh::Runners::Mesh

Includes:
Helpers::Lex
Included in:
Client, TaskRequest
Defined in:
lib/legion/extensions/mesh/runners/mesh.rb

Instance Method Summary collapse

Instance Method Details

#dispatch_gossip_message(type: nil, sender: nil, peers: []) ⇒ Object 



116
117
118
119
120
121
122
123
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 116

def dispatch_gossip_message(type: nil, sender: nil, peers: [], **)
  case type
  when 'mesh_gossip'
    merge_gossip(incoming_peers: peers, sender: sender)
  else
    { success: false, error: "unknown gossip message type: #{type}" }
  end
end

#expire_silent_agents ⇒ Object 



61
62
63
64
65
66
67
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 61

def expire_silent_agents(**)
  expired = mesh_registry.expire_silent_agents
  expired.each do |agent_id|
    log.info "[mesh] expired silent agent: #{agent_id}"
  end
  { success: true, expired: expired, count: expired.size }
end

#find_agents(capability:) ⇒ Object 



47
48
49
50
51
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 47

def find_agents(capability:, **)
  agents = mesh_registry.find_by_capability(capability)
  log.debug "[mesh] find: capability=#{capability} found=#{agents.size}"
  { agents: agents.map { |a| a[:agent_id] }, count: agents.size }
end

#heartbeat(agent_id:) ⇒ Object 



31
32
33
34
35
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 31

def heartbeat(agent_id:, **)
  result = mesh_registry.heartbeat(agent_id)
  log.debug "[mesh] heartbeat: agent=#{agent_id} alive=#{!result.nil?}"
  result ? { alive: true } : { error: :not_registered }
end

#merge_gossip(incoming_peers:, sender: nil) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 93

def merge_gossip(incoming_peers:, sender: nil, **)
  registry = mesh_registry
  conflict_agents = []
  merged = 0
  incoming_peers.each do |raw_peer|
    peer = raw_peer.transform_keys(&:to_sym)
    if peer[:node] == local_node_name
      conflict_agents << peer[:agent_id] if sender && sender != local_node_name
      next
    end
    merged += 1 if upsert_remote_peer(peer, registry)
  end

  if conflict_agents.any?
    publish_mesh_conflict(local_node: local_node_name, conflicting_node: sender,
                          conflict_agents: conflict_agents)
  end

  { success: true, merged: merged, total_peers: incoming_peers.size, conflicts: conflict_agents.size }
rescue StandardError => e
  { success: false, reason: :error, message: e.message }
end

#mesh_status ⇒ Object 



53
54
55
56
57
58
59
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 53

def mesh_status(**)
  online = mesh_registry.online_agents
  total = mesh_registry.count
  msgs = mesh_registry.messages.size
  log.debug "[mesh] status: total=#{total} online=#{online.size} messages=#{msgs}"
  { total: total, online: online.size, message_count: msgs }
end

#publish_gossip ⇒ Object 



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 69

def publish_gossip(**)
  expired = peer_table.expire
  registry = mesh_registry

  Array(expired).each do |agent_id|
    result = registry.unregister_agent(agent_id)
    if result
      log.info "[mesh] ttl-expired agent unregistered from registry: agent=#{agent_id}"
    else
      log.debug "[mesh] ttl-expired agent not found in registry: agent=#{agent_id}"
    end
  end
  peers = registry.all_agents.first(gossip_max_peers).map do |agent|
    agent.slice(:agent_id, :capabilities, :node, :source, :status, :generation,
                :last_seen, :registered_at).transform_values { |v| v.is_a?(Time) ? v.to_s : v }
  end

  @gossip_round = (@gossip_round || 0) + 1
  publish_gossip_message(peers)
  { success: true, peers_broadcast: peers.size, gossip_round: @gossip_round }
rescue StandardError => e
  { success: false, reason: :error, message: e.message }
end

#register(agent_id:, capabilities: [], endpoint: nil) ⇒ Object 



13
14
15
16
17
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 13

def register(agent_id:, capabilities: [], endpoint: nil, **)
  mesh_registry.register_agent(agent_id, capabilities: capabilities, endpoint: endpoint)
  log.info "[mesh] registered: agent=#{agent_id} capabilities=#{capabilities.join(',')}"
  { registered: true, agent_id: agent_id }
end

#send_message(from:, to: nil, capability: nil, pattern: :unicast, payload: {}) ⇒ Object 



37
38
39
40
41
42
43
44
45
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 37

def send_message(from:, to: nil, capability: nil, pattern: :unicast, payload: {}, **)
  return { error: :invalid_pattern } unless Helpers::Topology.valid_pattern?(pattern)

  msg = mesh_registry.route_message(from: from, to: to, capability: capability,
                                    pattern: pattern, payload: payload)
  count = msg[:delivered_to].size
  log.debug "[mesh] message: from=#{from} pattern=#{pattern} delivered=#{count} to=#{msg[:delivered_to].join(',')}"
  { sent: true, delivered_to: msg[:delivered_to], count: count }
end

#unregister(agent_id:) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/legion/extensions/mesh/runners/mesh.rb', line 19

def unregister(agent_id:, **)
  result = mesh_registry.unregister_agent(agent_id)
  if result
    log.info "[mesh] unregistered: agent=#{agent_id}"
    publish_mesh_departure(agent_id: agent_id, capabilities: result[:capabilities] || [])
    { unregistered: true }
  else
    log.debug "[mesh] unregister failed: agent=#{agent_id} not found"
    { error: :not_found }
  end
end