logos-messaging/logos-messaging-nim
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-nim.git synced 2026-08-25 15:31:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Clean separation between ReliableChannelManager, MessagingClient, and kernel/core (#3918)

Browse Source 
* Convert DeliveryService into optionally mountable MessagingClient
* Move SubscriptionManager to core layer (WakuNode)
* Ensure libwaku kernel_api/ still works (deprecated; removal pending)
* Create node_types.nim to allow WakuNode to compose subsystems cleanly
* Create node_telemetry.nim to centralize Prometheus types
* Remove unnecessary "ptr Waku" / "addr waku" indirection
* Rename Waku.startWaku -> Waku.start for upcoming Waku rename
* Write complete proc surface for SubscriptionManager (all intents expressible)
* Rename edgeFilterHealthLoop -> edgeFilterConnectionLoop ("Health" means monitoring)
* logosdelivery_start_node calls mountMessagingClient then starts
* libwaku and wakunode2 do not mount messagingClient
* Improve edge filter peer cleanup on disconnect
* misc refactors/moves, improvements, fixes

Co-authored-by: Ivan FB <128452529+Ivansete-status@users.noreply.github.com>
This commit is contained in:
Fabiana Cecin
2026-06-04 15:53:27 -03:00
committed by GitHub
co-authored by Ivan FB
parent 86e424c82c
commit f833ded209
37 changed files with 1454 additions and 1256 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/node/test_wakunode_health_monitor.nim
+10 -10
View File
@@ -15,8 +15,7 @@ import
node/health_monitor/protocol_health,
node/health_monitor/topic_health,
node/health_monitor/node_health_monitor,
node/delivery_service/delivery_service,
node/delivery_service/subscription_manager,
messaging_client,
node/kernel_api/relay,
node/kernel_api/store,
node/kernel_api/lightpush,
@@ -27,6 +26,7 @@ import
]
import ../testlib/[wakunode, wakucore], ../waku_archive/archive_utils
import waku/node/subscription_manager
const MockDLow = 4 # Mocked GossipSub DLow value
@@ -229,8 +229,8 @@ suite "Health Monitor - events":
await nodeA.start()
let ds =
DeliveryService.new(false, nodeA).expect("Failed to create DeliveryService")
ds.startDeliveryService().expect("Failed to start DeliveryService")
MessagingClient.new(false, nodeA).expect("Failed to create MessagingClient")
ds.start().expect("Failed to start MessagingClient")
let monitorA = NodeHealthMonitor.new(nodeA)
@@ -317,7 +317,7 @@ suite "Health Monitor - events":
lastStatus == ConnectionStatus.Disconnected
await monitorA.stopHealthMonitor()
await ds.stopDeliveryService()
await ds.stop()
await nodeA.stop()
asyncTest "Edge health driven by confirmed filter subscriptions":
@@ -333,9 +333,9 @@ suite "Health Monitor - events":
await nodeA.start()
let ds =
DeliveryService.new(false, nodeA).expect("Failed to create DeliveryService")
ds.startDeliveryService().expect("Failed to start DeliveryService")
let subMgr = ds.subscriptionManager
MessagingClient.new(false, nodeA).expect("Failed to create MessagingClient")
ds.start().expect("Failed to start MessagingClient")
let subMgr = nodeA.subscriptionManager
var nodeB: WakuNode
lockNewGlobalBrokerContext:
@@ -416,7 +416,7 @@ suite "Health Monitor - events":
await EventShardTopicHealthChange.dropListener(nodeA.brokerCtx, shardHealthLis)
check shardHealthOk == true
check subMgr.edgeFilterSubStates.len > 0
check nodeA.subscriptionManager.edgeFilterSubStates.len > 0
healthSignal.clear()
deadline = Moment.now() + TestConnectivityTimeLimit
@@ -428,7 +428,7 @@ suite "Health Monitor - events":
check lastStatus == ConnectionStatus.PartiallyConnected
await ds.stopDeliveryService()
await ds.stop()
await monitorA.stopHealthMonitor()
await nodeB.stop()
await nodeA.stop()
tests/node/test_wakunode_peer_exchange.nim
+102 -100
View File
@@ -9,7 +9,8 @@ import
libp2p/peerId,
libp2p/crypto/crypto,
eth/keys,
eth/p2p/discoveryv5/enr
eth/p2p/discoveryv5/enr,
brokers/broker_context
import
waku/[
@@ -184,114 +185,115 @@ suite "Waku Peer Exchange":
suite "Waku Peer Exchange with discv5":
asyncTest "Node successfully exchanges px peers with real discv5":
## Given (copied from test_waku_discv5.nim)
let
# todo: px flag
flags = CapabilitiesBitfield.init(
lightpush = false, filter = false, store = false, relay = true
)
bindIp = parseIpAddress("0.0.0.0")
extIp = parseIpAddress("127.0.0.1")
lockNewGlobalBrokerContext:
## Given (copied from test_waku_discv5.nim)
let
# todo: px flag
flags = CapabilitiesBitfield.init(
lightpush = false, filter = false, store = false, relay = true
)
bindIp = parseIpAddress("0.0.0.0")
extIp = parseIpAddress("127.0.0.1")
nodeKey1 = generateSecp256k1Key()
nodeTcpPort1 = Port(64010)
nodeUdpPort1 = Port(9000)
node1 = newTestWakuNode(
nodeKey1,
bindIp,
nodeTcpPort1,
some(extIp),
wakuFlags = some(flags),
discv5UdpPort = some(nodeUdpPort1),
nodeKey1 = generateSecp256k1Key()
nodeTcpPort1 = Port(64010)
nodeUdpPort1 = Port(9000)
node1 = newTestWakuNode(
nodeKey1,
bindIp,
nodeTcpPort1,
some(extIp),
wakuFlags = some(flags),
discv5UdpPort = some(nodeUdpPort1),
)
nodeKey2 = generateSecp256k1Key()
nodeTcpPort2 = Port(64012)
nodeUdpPort2 = Port(9002)
node2 = newTestWakuNode(
nodeKey2,
bindIp,
nodeTcpPort2,
some(extIp),
wakuFlags = some(flags),
discv5UdpPort = some(nodeUdpPort2),
)
nodeKey3 = generateSecp256k1Key()
nodeTcpPort3 = Port(64014)
nodeUdpPort3 = Port(9004)
node3 = newTestWakuNode(
nodeKey3,
bindIp,
nodeTcpPort3,
some(extIp),
wakuFlags = some(flags),
discv5UdpPort = some(nodeUdpPort3),
)
# discv5
let conf1 = WakuDiscoveryV5Config(
discv5Config: none(DiscoveryConfig),
address: bindIp,
port: nodeUdpPort1,
privateKey: keys.PrivateKey(nodeKey1.skkey),
bootstrapRecords: @[],
autoupdateRecord: true,
)
nodeKey2 = generateSecp256k1Key()
nodeTcpPort2 = Port(64012)
nodeUdpPort2 = Port(9002)
node2 = newTestWakuNode(
nodeKey2,
bindIp,
nodeTcpPort2,
some(extIp),
wakuFlags = some(flags),
discv5UdpPort = some(nodeUdpPort2),
let disc1 =
WakuDiscoveryV5.new(node1.rng, conf1, some(node1.enr), some(node1.peerManager))
let conf2 = WakuDiscoveryV5Config(
discv5Config: none(DiscoveryConfig),
address: bindIp,
port: nodeUdpPort2,
privateKey: keys.PrivateKey(nodeKey2.skkey),
bootstrapRecords: @[disc1.protocol.getRecord()],
autoupdateRecord: true,
)
nodeKey3 = generateSecp256k1Key()
nodeTcpPort3 = Port(64014)
nodeUdpPort3 = Port(9004)
node3 = newTestWakuNode(
nodeKey3,
bindIp,
nodeTcpPort3,
some(extIp),
wakuFlags = some(flags),
discv5UdpPort = some(nodeUdpPort3),
let disc2 =
WakuDiscoveryV5.new(node2.rng, conf2, some(node2.enr), some(node2.peerManager))
await allFutures(node1.start(), node2.start(), node3.start())
let resultDisc1StartRes = await disc1.start()
assert resultDisc1StartRes.isOk(), resultDisc1StartRes.error
let resultDisc2StartRes = await disc2.start()
assert resultDisc2StartRes.isOk(), resultDisc2StartRes.error
## When
var attempts = 10
while (disc1.protocol.nodesDiscovered < 1 or disc2.protocol.nodesDiscovered < 1) and
attempts > 0:
await sleepAsync(1.seconds)
attempts -= 1
# node2 can be connected, so will be returned by peer exchange
require (
await node1.peerManager.connectPeer(node2.switch.peerInfo.toRemotePeerInfo())
)
# discv5
let conf1 = WakuDiscoveryV5Config(
discv5Config: none(DiscoveryConfig),
address: bindIp,
port: nodeUdpPort1,
privateKey: keys.PrivateKey(nodeKey1.skkey),
bootstrapRecords: @[],
autoupdateRecord: true,
)
# Mount peer exchange
await node1.mountPeerExchange()
await node3.mountPeerExchange()
await node3.mountPeerExchangeClient()
let disc1 =
WakuDiscoveryV5.new(node1.rng, conf1, some(node1.enr), some(node1.peerManager))
let dialResponse =
await node3.dialForPeerExchange(node1.switch.peerInfo.toRemotePeerInfo())
let conf2 = WakuDiscoveryV5Config(
discv5Config: none(DiscoveryConfig),
address: bindIp,
port: nodeUdpPort2,
privateKey: keys.PrivateKey(nodeKey2.skkey),
bootstrapRecords: @[disc1.protocol.getRecord()],
autoupdateRecord: true,
)
check dialResponse.isOk
let disc2 =
WakuDiscoveryV5.new(node2.rng, conf2, some(node2.enr), some(node2.peerManager))
let
requestPeers = 1
currentPeers = node3.peerManager.switch.peerStore.peers.len
let res = await node3.fetchPeerExchangePeers(1)
check res.tryGet() == 1
await allFutures(node1.start(), node2.start(), node3.start())
let resultDisc1StartRes = await disc1.start()
assert resultDisc1StartRes.isOk(), resultDisc1StartRes.error
let resultDisc2StartRes = await disc2.start()
assert resultDisc2StartRes.isOk(), resultDisc2StartRes.error
# Then node3 has received 1 peer from node1
check:
node3.peerManager.switch.peerStore.peers.len == currentPeers + requestPeers
## When
var attempts = 10
while (disc1.protocol.nodesDiscovered < 1 or disc2.protocol.nodesDiscovered < 1) and
attempts > 0:
await sleepAsync(1.seconds)
attempts -= 1
# node2 can be connected, so will be returned by peer exchange
require (
await node1.peerManager.connectPeer(node2.switch.peerInfo.toRemotePeerInfo())
)
# Mount peer exchange
await node1.mountPeerExchange()
await node3.mountPeerExchange()
await node3.mountPeerExchangeClient()
let dialResponse =
await node3.dialForPeerExchange(node1.switch.peerInfo.toRemotePeerInfo())
check dialResponse.isOk
let
requestPeers = 1
currentPeers = node3.peerManager.switch.peerStore.peers.len
let res = await node3.fetchPeerExchangePeers(1)
check res.tryGet() == 1
# Then node3 has received 1 peer from node1
check:
node3.peerManager.switch.peerStore.peers.len == currentPeers + requestPeers
await allFutures(
[node1.stop(), node2.stop(), node3.stop(), disc1.stop(), disc2.stop()]
)
await allFutures(
[node1.stop(), node2.stop(), node3.stop(), disc1.stop(), disc2.stop()]
)