logos-messaging/logos-delivery
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-delivery.git synced 2026-07-19 02:59:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add fleet mode (#4019)

Browse Source 
* Rename WakuMode to LogosDeliveryMode, add Fleet mode
* KernelConf (which starts only kernel) is now a distinct WakuNodeConf (starts messaging, for tests)
* Add LogosDelivery.new(kernelConf) (kernel-only)
* Add LogosDelivery.new(kernelConf, messagingOverrides, channelsOverrides) (full stack)
* LogosDeliveryConf.messagingOverrides/channelsOverrides are Option and mount each layer if present
* Add kernel-only LogosDeliveryConf.init(kernelConf)
* Parse {"mode": "fleet", "kernelConf": {...}} in parseLogosDeliveryConf
* Generalize parseOverrides to parse over caller-supplied defaults
* Add requireMessaging/requireChannels FFI guards and apply them
* Make start/stop skip layers that were not mounted (no config)
* Rename toKernelConf to toWakuNodeConf
* Add tests to test_conf.nim
* Misc additions/refactors
This commit is contained in:
Fabiana Cecin 2026-07-09 15:57:16 -03:00 committed by GitHub
parent 90fa5fa91f
commit d83900aa9b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 317 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
library/channels_api/channel_api.nim
View File

@ -18,6 +18,9 @@ proc logosdelivery_channel_create(
requireInitializedNode(ctx, "ChannelCreate"):
return err(errMsg)
requireChannels(ctx, "ChannelCreate"):
return err(errMsg)
let id = ctx.myLib[].reliableChannelManager.createReliableChannel(
ChannelId($channelIdStr),
ContentTopic($contentTopicStr),
@ -38,6 +41,9 @@ proc logosdelivery_channel_send(
requireInitializedNode(ctx, "ChannelSend"):
return err(errMsg)
requireChannels(ctx, "ChannelSend"):
return err(errMsg)
var jsonNode: JsonNode
try:
jsonNode = parseJson($messageJson)
@ -70,6 +76,9 @@ proc logosdelivery_channel_close(
requireInitializedNode(ctx, "ChannelClose"):
return err(errMsg)
requireChannels(ctx, "ChannelClose"):
return err(errMsg)
(await ctx.myLib[].reliableChannelManager.closeChannel(ChannelId($channelIdStr))).isOkOr:
return err("ChannelClose failed: " & $error)

19
library/declare_lib.nim
View File

@ -1,5 +1,6 @@
import ffi
import std/locks
import results
import logos_delivery
declareLibrary("logosdelivery")
@ -17,6 +18,24 @@ template requireInitializedNode*(
let errMsg {.inject.} = opName & " failed: node is not initialized"
onError
template requireMessaging*(
ctx: ptr FFIContext[LogosDelivery], opName: string, onError: untyped
) =
## Use after `requireInitializedNode`. Fails if the node has no messaging client
## (a kernel-only / fleet node).
ctx.myLib[].ensureMessaging().isOkOr:
let errMsg {.inject.} = opName & " failed: " & error
onError
template requireChannels*(
ctx: ptr FFIContext[LogosDelivery], opName: string, onError: untyped
) =
## Use after `requireInitializedNode`. Fails if the node has no reliable channel
## manager (a kernel-only / fleet node).
ctx.myLib[].ensureChannels().isOkOr:
let errMsg {.inject.} = opName & " failed: " & error
onError
proc logosdelivery_set_event_callback(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.dynlib, exportc, cdecl.} =

9
library/logos_delivery_api/messaging_api.nim
View File

@ -17,6 +17,9 @@ proc logosdelivery_subscribe(
requireInitializedNode(ctx, "Subscribe"):
return err(errMsg)
requireMessaging(ctx, "Subscribe"):
return err(errMsg)
# ContentTopic is just a string type alias
let contentTopic = ContentTopic($contentTopicStr)
@ -35,6 +38,9 @@ proc logosdelivery_unsubscribe(
requireInitializedNode(ctx, "Unsubscribe"):
return err(errMsg)
requireMessaging(ctx, "Unsubscribe"):
return err(errMsg)
# ContentTopic is just a string type alias
let contentTopic = ContentTopic($contentTopicStr)
@ -53,6 +59,9 @@ proc logosdelivery_send(
requireInitializedNode(ctx, "Send"):
return err(errMsg)
requireMessaging(ctx, "Send"):
return err(errMsg)
## Parse the message JSON and send the message
var jsonNode: JsonNode
try:

4
logos_delivery/api/kernel_conf.nim
View File

@ -1,4 +1,6 @@
import tools/confutils/cli_args
export cli_args
type KernelConf* = WakuNodeConf ## Alias of the CLI-facing `WakuNodeConf`.
type KernelConf* = distinct WakuNodeConf
## Raw kernel config; distinct so `new(KernelConf)` doesn't collide with the
## full-stack `new(WakuNodeConf)`.

22
logos_delivery/api/logos_delivery_conf.nim
View File

@ -1,30 +1,38 @@
{.push raises: [].}
import std/options
import results
import logos_delivery/api/messaging_conf
import logos_delivery/channels/reliable_channel_manager
export messaging_conf, reliable_channel_manager
export options, messaging_conf, reliable_channel_manager
type LogosDeliveryConf* = object ## Aggregates the per-layer config objects.
type LogosDeliveryConf* = object
## Aggregates the per-layer config objects. A layer is mounted iff its config
## is present.
kernelConf*: KernelConf
messagingConf*: MessagingClientConf
channelsConf*: ReliableChannelManagerConf
messagingConf*: Option[MessagingClientConf]
channelsConf*: Option[ReliableChannelManagerConf]
proc init*(T: type LogosDeliveryConf, kernelConf: KernelConf): LogosDeliveryConf =
return LogosDeliveryConf(kernelConf: kernelConf)
proc init*(
T: type LogosDeliveryConf,
mode: WakuMode,
mode: LogosDeliveryMode,
preset: string,
messagingOverrides: MessagingClientConf,
channelsOverrides: ReliableChannelManagerConf,
): ConfResult[LogosDeliveryConf] =
let merged = merge(?resolvePreset(preset), messagingOverrides)
var kernelConf = ?toKernelConf(merged, mode)
var kernelConf = ?toWakuNodeConf(merged, mode)
kernelConf.preset = preset
return ok(
LogosDeliveryConf(
kernelConf: kernelConf, messagingConf: merged, channelsConf: channelsOverrides
kernelConf: KernelConf(kernelConf),
messagingConf: some(merged),
channelsConf: some(channelsOverrides),
)
)

46
logos_delivery/api/logos_delivery_conf_json.nim
View File

@ -10,23 +10,27 @@ const
# Lowercased, since `collectJsonFields` keys the object case-insensitively.
KeyMode = "mode"
KeyPreset = "preset"
KeyKernelConf = "kernelconf"
KeyMessagingOverrides = "messagingoverrides"
KeyChannelsOverrides = "channelsoverrides"
proc parseMode(s: string): Result[WakuMode, string] =
proc parseMode(s: string): Result[LogosDeliveryMode, string] =
case s.strip().toLowerAscii()
of "core":
return ok(WakuMode.Core)
return ok(LogosDeliveryMode.Core)
of "edge":
return ok(WakuMode.Edge)
return ok(LogosDeliveryMode.Edge)
of "fleet":
return ok(LogosDeliveryMode.Fleet)
else:
return err("invalid mode: '" & s & "' (expected 'Core' or 'Edge')")
return err("invalid mode: '" & s & "' (expected 'Core', 'Edge' or 'Fleet')")
proc parseOverrides[T](node: JsonNode, label: string): Result[T, string] =
proc parseOverrides[T](defaults: T, node: JsonNode, label: string): Result[T, string] =
## Parse the JSON object `node` as overrides on top of `defaults`.
if node.kind != JObject:
return err(label & " must be a JSON object")
var fields = ?collectJsonFields(node)
var conf = T()
var conf = defaults
?applyJsonFieldsToConf(
conf,
fields,
@ -45,11 +49,8 @@ proc parseLogosDeliveryConf*(jsonStr: string): ConfResult[LogosDeliveryConf] =
return err("configuration JSON must be an object")
var top = ?collectJsonFields(node)
var mode = WakuMode.Core
var preset = ""
var messagingOverrides = MessagingClientConf()
var channelsOverrides = ReliableChannelManagerConf()
var mode = LogosDeliveryMode.Core
if top.hasKey(KeyMode):
let (_, v) = top.getOrDefault(KeyMode)
if v.kind != JString:
@ -57,6 +58,22 @@ proc parseLogosDeliveryConf*(jsonStr: string): ConfResult[LogosDeliveryConf] =
mode = ?parseMode(v.getStr())
top.del(KeyMode)
if mode == LogosDeliveryMode.Fleet:
# Kernel-only: a raw kernelConf and no upper layers.
if not top.hasKey(KeyKernelConf):
return err("fleet mode requires a 'kernelConf' object")
let (_, v) = top.getOrDefault(KeyKernelConf)
let kernel = ?parseOverrides(?defaultWakuNodeConf(), v, "kernelConf")
top.del(KeyKernelConf)
if top.len > 0:
return
err(unknownKeysError(top, "fleet mode takes only 'kernelConf'; unexpected"))
return ok(LogosDeliveryConf.init(KernelConf(kernel)))
var preset = ""
var messagingOverrides = MessagingClientConf()
var channelsOverrides = ReliableChannelManagerConf()
if top.hasKey(KeyPreset):
let (_, v) = top.getOrDefault(KeyPreset)
if v.kind != JString:
@ -66,20 +83,17 @@ proc parseLogosDeliveryConf*(jsonStr: string): ConfResult[LogosDeliveryConf] =
if top.hasKey(KeyMessagingOverrides):
let (_, v) = top.getOrDefault(KeyMessagingOverrides)
messagingOverrides = ?parseOverrides[MessagingClientConf](v, "messagingOverrides")
messagingOverrides = ?parseOverrides(MessagingClientConf(), v, "messagingOverrides")
top.del(KeyMessagingOverrides)
if top.hasKey(KeyChannelsOverrides):
let (_, v) = top.getOrDefault(KeyChannelsOverrides)
channelsOverrides =
?parseOverrides[ReliableChannelManagerConf](v, "channelsOverrides")
?parseOverrides(ReliableChannelManagerConf(), v, "channelsOverrides")
top.del(KeyChannelsOverrides)
if top.len > 0:
var keys: seq[string]
for _, (k, _) in pairs(top):
keys.add(k)
return err("Unrecognized configuration option(s) found: " & keys.join(", "))
return err(unknownKeysError(top, "Unrecognized configuration option(s) found"))
return LogosDeliveryConf.init(mode, preset, messagingOverrides, channelsOverrides)

14
logos_delivery/api/messaging_conf.nim
View File

@ -8,28 +8,34 @@ import logos_delivery/waku/factory/networks_config
export kernel_conf, messaging_client
type WakuMode* {.pure.} = enum
type LogosDeliveryMode* {.pure.} = enum
Edge # client-only node
Core # full service node
Fleet # kernel-only node from a raw kernel config
proc toKernelConf*(self: MessagingClientConf, mode: WakuMode): ConfResult[KernelConf] =
proc toWakuNodeConf*(
self: MessagingClientConf, mode: LogosDeliveryMode
): ConfResult[WakuNodeConf] =
## Mode sets the protocol flags; set fields map to their kernel counterpart.
var conf = ?defaultWakuNodeConf()
case mode
of WakuMode.Core:
of LogosDeliveryMode.Core:
conf.relay = true
conf.filter = true
conf.lightpush = true
conf.discv5Discovery = some(true)
conf.peerExchange = true
conf.rendezvous = true
of WakuMode.Edge:
of LogosDeliveryMode.Edge:
conf.peerExchange = true
conf.relay = false
conf.filter = false
conf.lightpush = false
conf.store = false
of LogosDeliveryMode.Fleet:
return
err("fleet mode takes a raw kernel config; use LogosDelivery.new(kernelConf)")
if self.store.isSome():
conf.store = self.store.get()

86
logos_delivery/logos_delivery.nim
View File

@ -75,19 +75,26 @@ type LogosDelivery* = ref object ## Entry point. Holds one instance of each API
proc new*(
T: type LogosDelivery, conf: LogosDeliveryConf, appCallbacks: AppCallbacks = nil
): Future[Result[LogosDelivery, string]] {.async.} =
## Builds the stack bottom-up from a resolved per-layer config.
let wakuConf = conf.kernelConf.toWakuConf().valueOr:
return err("failed to handle the configuration: " & error)
## Builds the stack bottom-up from a resolved per-layer config; each layer is
## mounted iff its config is present.
let wakuConf = WakuNodeConf(conf.kernelConf).toWakuConf().valueOr:
return err("failed to handle the configuration: " & error)
let waku = (await Waku.new(wakuConf, appCallbacks)).valueOr:
return err("failed to create Waku: " & error)
let messagingClient = MessagingClient.new(conf.messagingConf, waku).valueOr:
return err("failed to create MessagingClient: " & error)
let messagingClient =
if conf.messagingConf.isSome():
MessagingClient.new(conf.messagingConf.get(), waku).valueOr:
return err("failed to create MessagingClient: " & error)
else:
nil
let reliableChannelManager = ReliableChannelManager.new(
conf.channelsConf, waku.brokerCtx
).valueOr:
return err("failed to create ReliableChannelManager: " & error)
let reliableChannelManager =
if conf.channelsConf.isSome():
ReliableChannelManager.new(conf.channelsConf.get(), waku.brokerCtx).valueOr:
return err("failed to create ReliableChannelManager: " & error)
else:
nil
return ok(
LogosDelivery(
@ -103,16 +110,41 @@ proc new*(
## Builds the full stack from a kernel `WakuNodeConf`.
return await LogosDelivery.new(
LogosDeliveryConf(
kernelConf: conf,
messagingConf: MessagingClientConf(),
channelsConf: ReliableChannelManagerConf(),
kernelConf: KernelConf(conf),
messagingConf: some(MessagingClientConf()),
channelsConf: some(ReliableChannelManagerConf()),
),
appCallbacks,
)
proc new*(
T: type LogosDelivery, kernelConf: KernelConf, appCallbacks: AppCallbacks = nil
): Future[Result[LogosDelivery, string]] {.async.} =
## Fleet mode: mounts the kernel only from a raw `KernelConf`; no messaging client,
## no channel manager.
return await LogosDelivery.new(LogosDeliveryConf.init(kernelConf), appCallbacks)
proc new*(
T: type LogosDelivery,
kernelConf: KernelConf,
messagingOverrides: MessagingClientConf,
channelsOverrides: ReliableChannelManagerConf,
appCallbacks: AppCallbacks = nil,
): Future[Result[LogosDelivery, string]] {.async.} =
## Full stack: kernel + messaging + channels. Messaging is never skipped; a
## kernel-only node uses `new(kernelConf)` instead.
return await LogosDelivery.new(
LogosDeliveryConf(
kernelConf: kernelConf,
messagingConf: some(messagingOverrides),
channelsConf: some(channelsOverrides),
),
appCallbacks,
)
proc new*(
T: type LogosDelivery,
mode: WakuMode = WakuMode.Core,
mode: LogosDeliveryMode = LogosDeliveryMode.Core,
preset: string = "",
messagingOverrides: MessagingClientConf = MessagingClientConf(),
channelsOverrides: ReliableChannelManagerConf = ReliableChannelManagerConf(),
@ -124,22 +156,20 @@ proc new*(
return await LogosDelivery.new(conf, appCallbacks)
proc start*(self: LogosDelivery): Future[Result[void, string]] {.async.} =
## Starts each layer bottom-up: transport first, then messaging, then channels.
## Starts each present layer bottom-up: transport, then messaging, then channels.
if self.waku.isNil():
return err("Waku node is not initialized")
if self.messagingClient.isNil():
return err("MessagingClient is not initialized")
if self.reliableChannelManager.isNil():
return err("ReliableChannelManager is not initialized")
(await self.waku.start()).isOkOr:
return err("failed to start Waku: " & error)
self.messagingClient.start().isOkOr:
return err("failed to start MessagingClient: " & error)
if not self.messagingClient.isNil():
self.messagingClient.start().isOkOr:
return err("failed to start MessagingClient: " & error)
self.reliableChannelManager.start().isOkOr:
return err("failed to start ReliableChannelManager: " & error)
if not self.reliableChannelManager.isNil():
self.reliableChannelManager.start().isOkOr:
return err("failed to start ReliableChannelManager: " & error)
return ok()
@ -160,6 +190,18 @@ proc isOnline*(self: LogosDelivery): Future[Result[bool, string]] {.async.} =
return err("Waku node is not initialized")
return await self.waku.isOnline()
proc ensureMessaging*(self: LogosDelivery): Result[void, string] =
## Fails if the node has no messaging client (a kernel-only / fleet node).
if self.isNil() or self.messagingClient.isNil():
return err("node has no messaging client (kernel-only/fleet node)")
ok()
proc ensureChannels*(self: LogosDelivery): Result[void, string] =
## Fails if the node has no reliable channel manager (a kernel-only / fleet node).
if self.isNil() or self.reliableChannelManager.isNil():
return err("node has no reliable channel manager (kernel-only/fleet node)")
ok()
# Compile-time check that each concrete type satisfies its API concept.
static:
doAssert Waku is KernelApi

4
tests/api/test_api_health.nim
View File

@ -91,7 +91,7 @@ suite "LM API health checking":
serviceNode.wakuRelay.subscribe(DefaultShard, dummyHandler)
lockNewGlobalBrokerContext:
var conf = MessagingClientConf().toKernelConf(Core).valueOr:
var conf = MessagingClientConf().toWakuNodeConf(Core).valueOr:
raiseAssert error
conf.listenAddress = parseIpAddress("0.0.0.0")
conf.tcpPort = Port(0)
@ -271,7 +271,7 @@ suite "LM API health checking":
var edgeWaku: LogosDelivery
lockNewGlobalBrokerContext:
var edgeConf = MessagingClientConf().toKernelConf(Edge).valueOr:
var edgeConf = MessagingClientConf().toWakuNodeConf(Edge).valueOr:
raiseAssert error
edgeConf.listenAddress = parseIpAddress("0.0.0.0")
edgeConf.tcpPort = Port(0)

3
tests/api/test_api_receive.nim
View File

@ -83,7 +83,8 @@ proc waitForConnectionStatus(
await EventConnectionStatusChange.dropListener(brokerCtx, handle)
proc createApiNodeConf(numShards: uint16 = 1): WakuNodeConf =
var conf = MessagingClientConf().toKernelConf(messaging_conf.WakuMode.Core).valueOr:
var conf = MessagingClientConf()
.toWakuNodeConf(messaging_conf.LogosDeliveryMode.Core).valueOr:
raiseAssert error
conf.listenAddress = parseIpAddress("0.0.0.0")
conf.tcpPort = Port(0)

22
tests/api/test_api_send.nim
View File

@ -121,9 +121,9 @@ proc validate(
check requestId == expectedRequestId
proc createApiNodeConf(
mode: messaging_conf.WakuMode = messaging_conf.WakuMode.Core
mode: messaging_conf.LogosDeliveryMode = messaging_conf.LogosDeliveryMode.Core
): WakuNodeConf =
var conf = MessagingClientConf().toKernelConf(mode).valueOr:
var conf = MessagingClientConf().toWakuNodeConf(mode).valueOr:
raiseAssert error
conf.listenAddress = parseIpAddress("0.0.0.0")
conf.tcpPort = Port(0)
@ -354,7 +354,11 @@ suite "Waku API - Send":
## connected to a lightpush-capable peer must deliver through lightpush.
var node: LogosDelivery
lockNewGlobalBrokerContext:
node = (await LogosDelivery.new(createApiNodeConf(messaging_conf.WakuMode.Edge))).valueOr:
node = (
await LogosDelivery.new(
createApiNodeConf(messaging_conf.LogosDeliveryMode.Edge)
)
).valueOr:
raiseAssert error
(await node.start()).isOkOr:
raiseAssert "Failed to start Waku node: " & error
@ -389,7 +393,11 @@ suite "Waku API - Send":
## later retry must deliver the queued message.
var node: LogosDelivery
lockNewGlobalBrokerContext:
node = (await LogosDelivery.new(createApiNodeConf(messaging_conf.WakuMode.Edge))).valueOr:
node = (
await LogosDelivery.new(
createApiNodeConf(messaging_conf.LogosDeliveryMode.Edge)
)
).valueOr:
raiseAssert error
(await node.start()).isOkOr:
raiseAssert "Failed to start Waku node: " & error
@ -480,7 +488,11 @@ suite "Waku API - Send":
var node: LogosDelivery
lockNewGlobalBrokerContext:
node = (await LogosDelivery.new(createApiNodeConf(messaging_conf.WakuMode.Edge))).valueOr:
node = (
await LogosDelivery.new(
createApiNodeConf(messaging_conf.LogosDeliveryMode.Edge)
)
).valueOr:
raiseAssert error
(await node.start()).isOkOr:
raiseAssert "Failed to start Waku node: " & error

26
tests/api/test_api_subscription.nim
View File

@ -69,9 +69,10 @@ type TestNetwork = ref object
publisherPeerInfo: RemotePeerInfo
proc createApiNodeConf(
mode: messaging_conf.WakuMode = messaging_conf.WakuMode.Core, numShards: uint16 = 1
mode: messaging_conf.LogosDeliveryMode = messaging_conf.LogosDeliveryMode.Core,
numShards: uint16 = 1,
): WakuNodeConf =
var conf = MessagingClientConf().toKernelConf(mode).valueOr:
var conf = MessagingClientConf().toWakuNodeConf(mode).valueOr:
raiseAssert error
conf.listenAddress = parseIpAddress("0.0.0.0")
conf.tcpPort = Port(0)
@ -89,7 +90,8 @@ proc setupSubscriberNode(conf: WakuNodeConf): Future[LogosDelivery] {.async.} =
return node
proc setupNetwork(
numShards: uint16 = 1, mode: messaging_conf.WakuMode = messaging_conf.WakuMode.Core
numShards: uint16 = 1,
mode: messaging_conf.LogosDeliveryMode = messaging_conf.LogosDeliveryMode.Core,
): Future[TestNetwork] {.async.} =
var net = TestNetwork()
@ -99,7 +101,7 @@ proc setupNetwork(
"Failed to mount metadata"
)
(await net.publisher.mountRelay()).expect("Failed to mount relay")
if mode == messaging_conf.WakuMode.Edge:
if mode == messaging_conf.LogosDeliveryMode.Edge:
await net.publisher.mountFilter()
await net.publisher.mountLibp2pPing()
await net.publisher.start()
@ -118,7 +120,7 @@ proc setupNetwork(
"Failed to sub publisher"
)
if mode == messaging_conf.WakuMode.Edge:
if mode == messaging_conf.LogosDeliveryMode.Edge:
lockNewGlobalBrokerContext:
net.meshBuddy = newTestWakuNode(generateSecp256k1Key())
net.meshBuddy.mountMetadata(3, toSeq(0'u16 ..< numShards)).expect(
@ -473,7 +475,7 @@ suite "Messaging API, SubscriptionManager":
await verifyNetworkState(activeSubs)
asyncTest "Subscription API, edge node subscribe and receive message":
let net = await setupNetwork(1, messaging_conf.WakuMode.Edge)
let net = await setupNetwork(1, messaging_conf.LogosDeliveryMode.Edge)
defer:
await net.teardown()
@ -495,7 +497,7 @@ suite "Messaging API, SubscriptionManager":
check eventManager.receivedMessages[0].contentTopic == testTopic
asyncTest "Subscription API, edge node ignores unsubscribed content topics":
let net = await setupNetwork(1, messaging_conf.WakuMode.Edge)
let net = await setupNetwork(1, messaging_conf.LogosDeliveryMode.Edge)
defer:
await net.teardown()
@ -517,7 +519,7 @@ suite "Messaging API, SubscriptionManager":
check eventManager.receivedMessages.len == 0
asyncTest "Subscription API, edge node unsubscribe stops message receipt":
let net = await setupNetwork(1, messaging_conf.WakuMode.Edge)
let net = await setupNetwork(1, messaging_conf.LogosDeliveryMode.Edge)
defer:
await net.teardown()
@ -542,7 +544,7 @@ suite "Messaging API, SubscriptionManager":
check eventManager.receivedMessages.len == 0
asyncTest "Subscription API, edge node overlapping topics isolation":
let net = await setupNetwork(1, messaging_conf.WakuMode.Edge)
let net = await setupNetwork(1, messaging_conf.LogosDeliveryMode.Edge)
defer:
await net.teardown()
@ -571,7 +573,7 @@ suite "Messaging API, SubscriptionManager":
check eventManager.receivedMessages[0].contentTopic == topicB
asyncTest "Subscription API, edge node resubscribe after unsubscribe":
let net = await setupNetwork(1, messaging_conf.WakuMode.Edge)
let net = await setupNetwork(1, messaging_conf.LogosDeliveryMode.Edge)
defer:
await net.teardown()
@ -656,7 +658,7 @@ suite "Messaging API, SubscriptionManager":
await meshBuddy.connectToNodes(@[publisherPeerInfo])
let conf = createApiNodeConf(messaging_conf.WakuMode.Edge, numShards)
let conf = createApiNodeConf(messaging_conf.LogosDeliveryMode.Edge, numShards)
var subscriber: LogosDelivery
lockNewGlobalBrokerContext:
subscriber =
@ -784,7 +786,7 @@ suite "Messaging API, SubscriptionManager":
await meshBuddy.connectToNodes(@[publisherPeerInfo])
await sparePeer.connectToNodes(@[publisherPeerInfo])
let conf = createApiNodeConf(messaging_conf.WakuMode.Edge, numShards)
let conf = createApiNodeConf(messaging_conf.LogosDeliveryMode.Edge, numShards)
var subscriber: LogosDelivery
lockNewGlobalBrokerContext:
subscriber =

151
tests/api/test_conf.nim
View File

@ -7,9 +7,9 @@ import logos_delivery/api/logos_delivery_conf_json
import logos_delivery/waku/factory/[waku_conf, networks_config]
import logos_delivery/waku/common/logging
suite "MessagingClientConf - mode expansion (toKernelConf)":
suite "MessagingClientConf - mode expansion (toWakuNodeConf)":
test "Core mode enables relay + service protocols":
let kc = MessagingClientConf().toKernelConf(WakuMode.Core).valueOr:
let kc = MessagingClientConf().toWakuNodeConf(LogosDeliveryMode.Core).valueOr:
raiseAssert error
check:
kc.relay == true
@ -20,7 +20,7 @@ suite "MessagingClientConf - mode expansion (toKernelConf)":
kc.rendezvous == true
test "Edge mode is client-only (no relay/filter/lightpush/store)":
let kc = MessagingClientConf().toKernelConf(WakuMode.Edge).valueOr:
let kc = MessagingClientConf().toWakuNodeConf(LogosDeliveryMode.Edge).valueOr:
raiseAssert error
check:
kc.relay == false
@ -37,7 +37,7 @@ suite "MessagingClientConf - field mapping + transport policy":
numShardsInCluster: some(4'u16),
maxMessageSize: some("150KiB"),
)
let kc = mc.toKernelConf(WakuMode.Core).valueOr:
let kc = mc.toWakuNodeConf(LogosDeliveryMode.Core).valueOr:
raiseAssert error
check:
kc.clusterId == some(3'u16)
@ -45,7 +45,7 @@ suite "MessagingClientConf - field mapping + transport policy":
kc.maxMessageSize == "150KiB"
test "messaging transport defaults: ephemeral ports, websocket off, quic on":
let kc = MessagingClientConf().toKernelConf(WakuMode.Core).valueOr:
let kc = MessagingClientConf().toWakuNodeConf(LogosDeliveryMode.Core).valueOr:
raiseAssert error
check:
kc.tcpPort == Port(0)
@ -59,7 +59,7 @@ suite "MessagingClientConf - field mapping + transport policy":
websocketSupport: some(true),
quicSupport: some(false),
)
let kc = mc.toKernelConf(WakuMode.Core).valueOr:
let kc = mc.toWakuNodeConf(LogosDeliveryMode.Core).valueOr:
raiseAssert error
check:
kc.tcpPort == Port(1234)
@ -91,7 +91,7 @@ suite "MessagingClientConf - preset resolution":
let presetConf = resolvePreset("logos.dev").valueOr:
raiseAssert error
let merged = merge(presetConf, MessagingClientConf(numShardsInCluster: some(1'u16)))
var kernelConf = toKernelConf(merged, WakuMode.Core).valueOr:
var kernelConf = toWakuNodeConf(merged, LogosDeliveryMode.Core).valueOr:
raiseAssert error
kernelConf.preset = "logos.dev"
let wakuConf = kernelConf.toWakuConf().valueOr:
@ -110,38 +110,59 @@ suite "MessagingClientConf - merge (override wins)":
mc.maxMessageSize == some("1MB") # base preserved
suite "parseLogosDeliveryConf - JSON parsing":
test "empty object -> Core (default), empty preset, empty overrides":
test "empty object resolves to a full Core node conf":
let lc = parseLogosDeliveryConf("{}").valueOr:
raiseAssert error
check:
lc.kernelConf.relay == true # Core enables relay
lc.kernelConf.preset == ""
lc.messagingConf.clusterId.isNone()
lc.messagingConf.isSome() # full node: messaging + channels mounted
lc.channelsConf.isSome()
WakuNodeConf(lc.kernelConf).relay == true # Core enables relay
WakuNodeConf(lc.kernelConf).clusterId.isNone() # no cluster set
test "mode + preset are parsed":
test "mode + preset fold into the kernel conf":
let lc = parseLogosDeliveryConf("""{"mode": "Edge", "preset": "logostest"}""").valueOr:
raiseAssert error
let kc = WakuNodeConf(lc.kernelConf)
check:
lc.kernelConf.preset == "logostest"
lc.kernelConf.relay == false # Edge disables relay
kc.preset == "logostest"
kc.relay == false # Edge disables relay
test "messagingOverrides parsed into the partial":
test "messaging overrides split: kernel fields to the kernel, reliability to the record":
let lc = parseLogosDeliveryConf(
"""{"mode": "Core", "messagingOverrides": {"clusterId": 7, "reliabilityEnabled": true}}"""
).valueOr:
raiseAssert error
require lc.messagingConf.isSome()
check:
lc.messagingConf.clusterId == some(7'u16)
lc.messagingConf.reliabilityEnabled == some(true)
WakuNodeConf(lc.kernelConf).clusterId == some(7'u16) # kernel field
lc.messagingConf.get().reliabilityEnabled == some(true) # messaging-only
test "channelsOverrides parsed into the partial":
test "messaging overrides are recorded verbatim, unset fields left none":
let lc = parseLogosDeliveryConf("""{"messagingOverrides": {"clusterId": 7}}""").valueOr:
raiseAssert error
require lc.messagingConf.isSome()
let overrides = lc.messagingConf.get()
check:
overrides.clusterId == some(7'u16) # the user's override, kept as given
overrides.reliabilityEnabled.isNone() # user didn't set it, so it stays none
test "a preset resolves reliability into the messaging record":
let lc = parseLogosDeliveryConf("""{"preset": "twn"}""").valueOr:
raiseAssert error
# the user set no reliability override, but the preset supplies one
require lc.messagingConf.isSome()
check lc.messagingConf.get().reliabilityEnabled.isSome()
test "channelsOverrides fold into the channel conf":
let lc = parseLogosDeliveryConf(
"""{"channelsOverrides": {"rateLimitEnabled": true, "sdsMaxRetransmissions": 9}}"""
).valueOr:
raiseAssert error
require lc.channelsConf.isSome()
let channels = lc.channelsConf.get()
check:
lc.channelsConf.rateLimitEnabled == some(true)
lc.channelsConf.sdsMaxRetransmissions == some(9)
channels.rateLimitEnabled == some(true)
channels.sdsMaxRetransmissions == some(9)
test "invalid mode is rejected (Core or Edge only)":
check parseLogosDeliveryConf("""{"mode": "bogus"}""").isErr()
@ -159,10 +180,12 @@ suite "parseLogosDeliveryConf - JSON parsing":
"""{"messagingOverrides": {"cluster-id": 7, "reliability": true, "tcp-port": 1234}}"""
).valueOr:
raiseAssert error
let kc = WakuNodeConf(lc.kernelConf)
require lc.messagingConf.isSome()
check:
lc.messagingConf.clusterId == some(7'u16)
lc.messagingConf.reliabilityEnabled == some(true)
lc.messagingConf.p2pTcpPort == some(Port(1234))
kc.clusterId == some(7'u16)
lc.messagingConf.get().reliabilityEnabled == some(true) # messaging-only
kc.tcpPort == Port(1234)
test "unknown keys inside an overrides body are rejected":
check parseLogosDeliveryConf("""{"messagingOverrides": {"bogusKey": 1}}""").isErr()
@ -179,19 +202,21 @@ suite "parseLogosDeliveryConf - JSON parsing":
"""{"messagingOverrides": {"log-level": "DEBUG", "log-format": "JSON", "nodekey": "0d714a1fada214dead6dc9c7274581ec20ff292451866e7d6d677dc818e8ccd2"}}"""
).valueOr:
raiseAssert error
let kc = WakuNodeConf(lc.kernelConf)
check:
lc.kernelConf.logLevel == logging.LogLevel.DEBUG
lc.kernelConf.logFormat == logging.LogFormat.JSON
lc.kernelConf.nodekey.isSome()
kc.logLevel == logging.LogLevel.DEBUG
kc.logFormat == logging.LogFormat.JSON
kc.nodekey.isSome()
test "a null value leaves the field unset":
let lc = parseLogosDeliveryConf(
"""{"messagingOverrides": {"store": null, "clusterId": 7}}"""
).valueOr:
raiseAssert error
let kc = WakuNodeConf(lc.kernelConf)
check:
lc.messagingConf.store.isNone()
lc.messagingConf.clusterId == some(7'u16)
kc.clusterId == some(7'u16)
kc.store == false # null left store unset; Core does not enable it
test "an invalid Ethereum RPC URL is rejected at parse time":
check parseLogosDeliveryConf(
@ -202,18 +227,50 @@ suite "parseLogosDeliveryConf - JSON parsing":
"""{"messagingOverrides": {"rln-relay-eth-client-address": ["http://localhost:8540/"]}}"""
).valueOr:
raiseAssert error
check lc.kernelConf.ethClientUrls.len == 1
check WakuNodeConf(lc.kernelConf).ethClientUrls.len == 1
test "store backend fields parse and map to the kernel":
test "store backend fields fold into the kernel conf":
let lc = parseLogosDeliveryConf(
"""{"messagingOverrides": {"store": true, "store-message-db-url": "sqlite://test.db", "store-message-retention-policy": "time:3600", "store-max-num-db-connections": 7}}"""
).valueOr:
raiseAssert error
let kc = WakuNodeConf(lc.kernelConf)
check:
lc.kernelConf.store == true
lc.kernelConf.storeMessageDbUrl == "sqlite://test.db"
lc.kernelConf.storeMessageRetentionPolicy == "time:3600"
lc.kernelConf.storeMaxNumDbConnections == 7
kc.store == true
kc.storeMessageDbUrl == "sqlite://test.db"
kc.storeMessageRetentionPolicy == "time:3600"
kc.storeMaxNumDbConnections == 7
test "fleet mode parses a raw kernelConf":
let lc = parseLogosDeliveryConf(
"""{"mode": "fleet", "kernelConf": {"relay": false, "maxMessageSize": "150KiB"}}"""
).valueOr:
raiseAssert error
check:
lc.messagingConf.isNone() # kernel-only: no upper layers
lc.channelsConf.isNone()
let kc = WakuNodeConf(lc.kernelConf)
check:
kc.relay == false
kc.maxMessageSize == "150KiB"
test "fleet mode requires a kernelConf":
check parseLogosDeliveryConf("""{"mode": "fleet"}""").isErr()
test "fleet mode rejects anything besides kernelConf":
check parseLogosDeliveryConf(
"""{"mode": "fleet", "kernelConf": {}, "messagingOverrides": {"clusterId": 1}}"""
)
.isErr()
# a preset for a fleet node goes inside kernelConf (WakuNodeConf.preset); fleet
# treats kernelConf as a finished object and overlays nothing onto it
check parseLogosDeliveryConf(
"""{"mode": "fleet", "kernelConf": {}, "preset": "twn"}"""
)
.isErr()
test "kernelConf is rejected outside fleet mode":
check parseLogosDeliveryConf("""{"mode": "core", "kernelConf": {}}""").isErr()
suite "LogosDelivery.new - construction (the app-dev entry)":
asyncTest "builds the full messaging stack from mode + overrides":
@ -221,7 +278,7 @@ suite "LogosDelivery.new - construction (the app-dev entry)":
lockNewGlobalBrokerContext:
node = (
await LogosDelivery.new(
WakuMode.Core,
LogosDeliveryMode.Core,
"",
MessagingClientConf(
clusterId: some(3'u16),
@ -243,7 +300,7 @@ suite "LogosDelivery.new - construction (the app-dev entry)":
lockNewGlobalBrokerContext:
node = (
await LogosDelivery.new(
WakuMode.Core,
LogosDeliveryMode.Core,
"logostest",
MessagingClientConf(listenIpv4: some(parseIpAddress("0.0.0.0"))),
)
@ -254,8 +311,28 @@ suite "LogosDelivery.new - construction (the app-dev entry)":
suite "MessagingClientConf - store override":
test "store opt-in overrides the mode default; protocol flags follow the mode":
let kc = MessagingClientConf(store: some(true)).toKernelConf(WakuMode.Edge).valueOr:
let kc = MessagingClientConf(store: some(true)).toWakuNodeConf(
LogosDeliveryMode.Edge
).valueOr:
raiseAssert error
check:
kc.store == true # Edge defaults store off; the explicit opt-in wins
kc.relay == false # protocols are owned by the mode, not overridable
suite "LogosDelivery.new - raw kernel construction":
asyncTest "a fleet node mounts the kernel only; start/stop tolerate the nil layers":
let kernel = MessagingClientConf(listenIpv4: some(parseIpAddress("0.0.0.0"))).toWakuNodeConf(
LogosDeliveryMode.Core
).valueOr:
raiseAssert error
var node: LogosDelivery
lockNewGlobalBrokerContext:
node = (await LogosDelivery.new(KernelConf(kernel))).valueOr:
raiseAssert error
check:
not node.waku.isNil()
node.messagingClient.isNil() # no messaging client on a fleet node
node.reliableChannelManager.isNil() # no channel manager either
# start()/stop() must skip the nil messaging + channel layers, not deref them
(await node.start()).expect("start")
(await node.stop()).expect("stop")

3
tests/channels/test_reliable_channel_send_receive.nim
View File

@ -28,7 +28,8 @@ import snapshot_codec
const TestTimeout = chronos.seconds(15)
proc createApiNodeConf(): WakuNodeConf =
var conf = MessagingClientConf().toKernelConf(messaging_conf.WakuMode.Core).valueOr:
var conf = MessagingClientConf()
.toWakuNodeConf(messaging_conf.LogosDeliveryMode.Core).valueOr:
raiseAssert error
conf.listenAddress = parseIpAddress("0.0.0.0")
conf.tcpPort = Port(0)

12
tests/test_waku.nim
View File

@ -12,8 +12,8 @@ import logos_delivery/waku/factory/conf_builder/conf_builder
suite "LogosDelivery API - Create node":
asyncTest "Create node with minimal configuration":
## Given
var nodeConf = MessagingClientConf().toKernelConf(Core).valueOr:
raiseAssert "toKernelConf failed: " & error
var nodeConf = MessagingClientConf().toWakuNodeConf(Core).valueOr:
raiseAssert "toWakuNodeConf failed: " & error
nodeConf.clusterId = some(3'u16)
nodeConf.rest = false
@ -31,8 +31,8 @@ suite "LogosDelivery API - Create node":
asyncTest "Create node with full configuration":
## Given
var nodeConf = MessagingClientConf().toKernelConf(Core).valueOr:
raiseAssert "toKernelConf failed: " & error
var nodeConf = MessagingClientConf().toWakuNodeConf(Core).valueOr:
raiseAssert "toWakuNodeConf failed: " & error
nodeConf.clusterId = some(99'u16)
nodeConf.rest = false
nodeConf.numShardsInNetwork = 16
@ -62,8 +62,8 @@ suite "LogosDelivery API - Create node":
asyncTest "Create node with mixed entry nodes (enrtree, multiaddr)":
## Given
var nodeConf = MessagingClientConf().toKernelConf(Core).valueOr:
raiseAssert "toKernelConf failed: " & error
var nodeConf = MessagingClientConf().toWakuNodeConf(Core).valueOr:
raiseAssert "toWakuNodeConf failed: " & error
nodeConf.clusterId = some(42'u16)
nodeConf.rest = false
nodeConf.entryNodes = @[

2
tools/confutils/conf_from_json.nim
View File

@ -23,7 +23,7 @@ proc collectJsonFields*(
jsonFields[lowerKey] = (key, value)
return ok(jsonFields)
proc unknownKeysError(
proc unknownKeysError*(
jsonFields: Table[string, (string, JsonNode)], prefix: string
): string =
## Format leftover JSON keys as an error message.