diff --git a/logos_delivery/api/conf/logos_delivery_conf.nim b/logos_delivery/api/conf/logos_delivery_conf.nim index bec9f9a78..eb2e5b9fa 100644 --- a/logos_delivery/api/conf/logos_delivery_conf.nim +++ b/logos_delivery/api/conf/logos_delivery_conf.nim @@ -2,10 +2,11 @@ import results +import logos_delivery/api/conf/modes import logos_delivery/api/conf/messaging_conf import logos_delivery/api/conf/channels_conf -export messaging_conf, channels_conf +export modes, messaging_conf, channels_conf type LogosDeliveryConf* = object ## Aggregates the per-layer config objects. A layer is mounted iff its config @@ -19,11 +20,14 @@ proc init*(T: type LogosDeliveryConf, kernelConf: KernelConf): LogosDeliveryConf proc init*( T: type LogosDeliveryConf, + entryLayer: EntryLayer = EntryLayer.channels, mode: LogosDeliveryMode, preset: string, messagingOverrides: MessagingClientConf, channelsOverrides: ReliableChannelManagerConf, ): ConfResult[LogosDeliveryConf] = + ## Structured (preset + overrides) entry. Only `messaging` / `channels` layers + ## reach here; the `kernel` layer uses `init(kernelConf)` (raw, mode ignored). let merged = merge(?resolvePreset(preset), messagingOverrides) var kernelConf = ?toWakuNodeConf(merged, mode) kernelConf.preset = preset @@ -31,7 +35,11 @@ proc init*( LogosDeliveryConf( kernelConf: KernelConf(kernelConf), messagingConf: Opt.some(merged), - channelsConf: Opt.some(channelsOverrides), + channelsConf: + if entryLayer == EntryLayer.channels: + Opt.some(channelsOverrides) + else: + Opt.none(ReliableChannelManagerConf), ) ) diff --git a/logos_delivery/api/conf/logos_delivery_conf_json.nim b/logos_delivery/api/conf/logos_delivery_conf_json.nim index 6440a63ac..307ca51c6 100644 --- a/logos_delivery/api/conf/logos_delivery_conf_json.nim +++ b/logos_delivery/api/conf/logos_delivery_conf_json.nim @@ -8,6 +8,7 @@ import logos_delivery/api/conf/logos_delivery_conf const # Lowercased, since `collectJsonFields` keys the object case-insensitively. + KeyEntryLayer = "entrylayer" KeyMode = "mode" KeyPreset = "preset" KeyKernelConf = "kernelconf" @@ -24,10 +25,21 @@ proc parseMode(s: string): Result[LogosDeliveryMode, string] = return ok(LogosDeliveryMode.Core) of "edge": return ok(LogosDeliveryMode.Edge) - of "fleet": - return ok(LogosDeliveryMode.Fleet) else: - return err("invalid mode: '" & s & "' (expected 'Core', 'Edge' or 'Fleet')") + return err("invalid mode: '" & s & "' (expected 'Core' or 'Edge')") + +proc parseEntryLayer(s: string): Result[EntryLayer, string] = + case s.strip().toLowerAscii() + of "kernel": + return ok(EntryLayer.kernel) + of "messaging": + return ok(EntryLayer.messaging) + of "channels": + return ok(EntryLayer.channels) + else: + return err( + "invalid entryLayer: '" & s & "' (expected 'kernel', 'messaging' or 'channels')" + ) proc parseOverrides[T](defaults: T, node: JsonNode, label: string): Result[T, string] = ## Parse the JSON object `node` as overrides on top of `defaults`. @@ -106,16 +118,25 @@ 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. + var entryLayer = EntryLayer.channels + if top.hasKey(KeyEntryLayer): + let (_, v) = top.getOrDefault(KeyEntryLayer) + if v.kind != JString: + return err("entryLayer must be a string") + entryLayer = ?parseEntryLayer(v.getStr()) + top.del(KeyEntryLayer) + + if entryLayer == EntryLayer.kernel: + # Kernel-only: a raw kernelConf and no upper layers; mode is ignored. if not top.hasKey(KeyKernelConf): - return err("fleet mode requires a 'kernelConf' object") + return err("kernel entry layer 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 err( + unknownKeysError(top, "kernel entry layer takes only 'kernelConf'; unexpected") + ) return ok(LogosDeliveryConf.init(KernelConf(kernel))) # [Legacy flat JSON config] A wrapper key marks our structured shape. Otherwise any @@ -158,6 +179,12 @@ proc parseLogosDeliveryConf*(jsonStr: string): ConfResult[LogosDeliveryConf] = if top.len > 0: return err(unknownKeysError(top, "Unrecognized configuration option(s) found")) - return LogosDeliveryConf.init(mode, preset, messagingOverrides, channelsOverrides) + return LogosDeliveryConf.init( + entryLayer = entryLayer, + mode = mode, + preset = preset, + messagingOverrides = messagingOverrides, + channelsOverrides = channelsOverrides, + ) {.pop.} diff --git a/logos_delivery/api/conf/messaging_conf.nim b/logos_delivery/api/conf/messaging_conf.nim index a8c91acd7..32e1f1542 100644 --- a/logos_delivery/api/conf/messaging_conf.nim +++ b/logos_delivery/api/conf/messaging_conf.nim @@ -7,10 +7,8 @@ import logos_delivery/waku/factory/networks_config export kernel_conf -type LogosDeliveryMode* {.pure.} = enum - Edge # client-only node - Core # full service node - Fleet # kernel-only node from a raw kernel config +# `LogosDeliveryMode` and `EntryLayer` are defined at the leaf (`cli_args`) so +# they can appear on `WakuNodeConf`; re-exported here via `kernel_conf`. type MessagingClientConf* = object clusterId* {.name: "cluster-id".}: Opt[uint16] ## Network cluster id. @@ -69,9 +67,6 @@ proc applyMode*(conf: var WakuNodeConf, mode: LogosDeliveryMode): ConfResult[voi 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)") return ok() proc toWakuNodeConf*( diff --git a/logos_delivery/api/conf/modes.nim b/logos_delivery/api/conf/modes.nim new file mode 100644 index 000000000..56058aed9 --- /dev/null +++ b/logos_delivery/api/conf/modes.nim @@ -0,0 +1,18 @@ +## Leaf module for the app-level mode / entry-layer enums. +## +## These appear on `WakuNodeConf` (in the leaf `tools/confutils/cli_args`), so +## they must live in a module that `cli_args` can import without a cycle — i.e. +## a module that imports nothing from the config/api layers. `logos_delivery_conf` +## re-exports them so consumers still get them from there. + +type LogosDeliveryMode* {.pure.} = enum + ## Drives the kernel-internal protocol mountings. Applied only for the + ## `messaging` / `channels` entry layers; ignored when `entryLayer == kernel`. + Edge # client-only node + Core # full service node + +type EntryLayer* {.pure.} = enum + ## Selects which API layer `LogosDelivery` instantiates. + kernel # transport kernel only; ignores `mode` and uses the config as-is + messaging # kernel + messaging client + channels # kernel + messaging + reliable channels diff --git a/logos_delivery/logos_delivery.nim b/logos_delivery/logos_delivery.nim index 463b0b343..55867e582 100644 --- a/logos_delivery/logos_delivery.nim +++ b/logos_delivery/logos_delivery.nim @@ -108,21 +108,38 @@ proc new*( proc new*( T: type LogosDelivery, conf: WakuNodeConf, appCallbacks: AppCallbacks = nil ): Future[Result[LogosDelivery, string]] {.async.} = - ## Builds the full stack from a kernel `WakuNodeConf`. - return await LogosDelivery.new( - LogosDeliveryConf( - kernelConf: KernelConf(conf), - messagingConf: Opt.some(MessagingClientConf()), - channelsConf: Opt.some(ReliableChannelManagerConf()), - ), - appCallbacks, + ## Builds the stack from a kernel `WakuNodeConf`, selecting which API layers to + ## instantiate by `conf.entryLayer`: + ## kernel -> transport only; `conf.mode` is ignored and the config is used as-is + ## messaging -> kernel + messaging client + ## channels -> kernel + messaging + reliable channels + ## For `messaging`/`channels`, `conf.mode` (Edge/Core) sets the kernel protocol + ## flags first (messaging-level concern); for `kernel` it is skipped. + var kernelConf = conf + if conf.entryLayer != EntryLayer.kernel: + applyMode(kernelConf, conf.mode).isOkOr: + return err("failed to apply mode: " & error) + + let ldConf = LogosDeliveryConf( + kernelConf: KernelConf(kernelConf), + messagingConf: + if conf.entryLayer == EntryLayer.kernel: + Opt.none(MessagingClientConf) + else: + Opt.some(MessagingClientConf()), + channelsConf: + if conf.entryLayer == EntryLayer.channels: + Opt.some(ReliableChannelManagerConf()) + else: + Opt.none(ReliableChannelManagerConf), ) + return await LogosDelivery.new(ldConf, 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. + ## Kernel entry layer: mounts the kernel only from a raw `KernelConf`; no + ## messaging client, no channel manager. return await LogosDelivery.new(LogosDeliveryConf.init(kernelConf), appCallbacks) proc new*( @@ -145,14 +162,23 @@ proc new*( proc new*( T: type LogosDelivery, + entryLayer: EntryLayer = EntryLayer.channels, mode: LogosDeliveryMode = LogosDeliveryMode.Core, preset: string = "", messagingOverrides: MessagingClientConf = MessagingClientConf(), channelsOverrides: ReliableChannelManagerConf = ReliableChannelManagerConf(), appCallbacks: AppCallbacks = nil, ): Future[Result[LogosDelivery, string]] {.async.} = - ## Messaging entry point (app dev). Builds the full stack from preset, mode and overrides. - let conf = LogosDeliveryConf.init(mode, preset, messagingOverrides, channelsOverrides).valueOr: + ## Messaging entry point (app dev). Builds the stack from preset, mode and + ## overrides; `entryLayer` selects messaging vs channels (use `new(kernelConf)` + ## for a kernel-only node). + let conf = LogosDeliveryConf.init( + entryLayer = entryLayer, + mode = mode, + preset = preset, + messagingOverrides = messagingOverrides, + channelsOverrides = channelsOverrides, + ).valueOr: return err("failed to synthesize configuration: " & error) return await LogosDelivery.new(conf, appCallbacks) @@ -196,15 +222,15 @@ proc isOnline*(self: LogosDelivery): Future[Result[bool, string]] {.async.} = 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). + ## Fails if the node has no messaging client (a kernel-only node). if self.isNil() or self.messagingClient.isNil(): - return err("node has no messaging client (kernel-only/fleet node)") + return err("node has no messaging client (kernel-only node)") ok() proc ensureChannels*(self: LogosDelivery): Result[void, string] = - ## Fails if the node has no reliable channel manager (a kernel-only / fleet node). + ## Fails if the node has no reliable channel manager (a kernel-only node). if self.isNil() or self.reliableChannelManager.isNil(): - return err("node has no reliable channel manager (kernel-only/fleet node)") + return err("node has no reliable channel manager (kernel-only node)") ok() # Compile-time check that each concrete type satisfies its API concept. diff --git a/tests/api/test_all.nim b/tests/api/test_all.nim index 22af4748d..3d7bdee4b 100644 --- a/tests/api/test_all.nim +++ b/tests/api/test_all.nim @@ -8,4 +8,5 @@ import ./test_api_subscription, ./test_api_receive, ./test_api_health, - ./test_messaging_rest + ./test_messaging_rest, + ./test_entry_layer diff --git a/tests/api/test_entry_layer.nim b/tests/api/test_entry_layer.nim new file mode 100644 index 000000000..12fc92236 --- /dev/null +++ b/tests/api/test_entry_layer.nim @@ -0,0 +1,73 @@ +{.used.} + +import std/[options, net] +import chronos, testutils/unittests +import brokers/broker_context +import logos_delivery +import logos_delivery/api/conf/logos_delivery_conf +import tools/confutils/cli_args +import ../testlib/testasync + +## Validates the layer-selection invariant of `LogosDelivery.new(WakuNodeConf)`: +## `messagingClient` (and `reliableChannelManager`) are instantiated only for the +## entry layers that call for them. +## +## kernel -> waku only +## messaging -> waku + messagingClient +## channels -> waku + messagingClient + reliableChannelManager + +proc nodeConf(entryLayer: EntryLayer): WakuNodeConf = + var conf = defaultWakuNodeConf().valueOr: + raiseAssert error + conf.entryLayer = entryLayer + conf.mode = LogosDeliveryMode.Core + conf.listenAddress = parseIpAddress("0.0.0.0") + conf.tcpPort = Port(0) + conf.discv5UdpPort = Port(0) + conf.clusterId = some(3'u16) + conf.numShardsInNetwork = 1 + conf.rest = false + return conf + +suite "LogosDelivery - entry layer selection": + asyncTest "kernel: waku only, no messaging / channels": + var node: LogosDelivery + lockNewGlobalBrokerContext: + node = (await LogosDelivery.new(nodeConf(EntryLayer.kernel))).valueOr: + raiseAssert error + check: + not node.waku.isNil() + node.messagingClient.isNil() + node.reliableChannelManager.isNil() + node.ensureMessaging().isErr() + node.ensureChannels().isErr() + (await node.stop()).isOkOr: + raiseAssert "stop failed: " & error + + asyncTest "messaging: waku + messagingClient, no channels": + var node: LogosDelivery + lockNewGlobalBrokerContext: + node = (await LogosDelivery.new(nodeConf(EntryLayer.messaging))).valueOr: + raiseAssert error + check: + not node.waku.isNil() + not node.messagingClient.isNil() + node.reliableChannelManager.isNil() + node.ensureMessaging().isOk() + node.ensureChannels().isErr() + (await node.stop()).isOkOr: + raiseAssert "stop failed: " & error + + asyncTest "channels: full stack": + var node: LogosDelivery + lockNewGlobalBrokerContext: + node = (await LogosDelivery.new(nodeConf(EntryLayer.channels))).valueOr: + raiseAssert error + check: + not node.waku.isNil() + not node.messagingClient.isNil() + not node.reliableChannelManager.isNil() + node.ensureMessaging().isOk() + node.ensureChannels().isOk() + (await node.stop()).isOkOr: + raiseAssert "stop failed: " & error diff --git a/tests/api/test_messaging_rest.nim b/tests/api/test_messaging_rest.nim index 66c1816d8..91ebaf349 100644 --- a/tests/api/test_messaging_rest.nim +++ b/tests/api/test_messaging_rest.nim @@ -10,6 +10,7 @@ import import brokers/broker_context import logos_delivery import + logos_delivery/api/conf/logos_delivery_conf, logos_delivery/messaging/rest_api/client as messaging_rest_client, logos_delivery/waku/rest_api/endpoint/client, logos_delivery/waku/common/base64 @@ -28,13 +29,13 @@ import ../testlib/[wakucore, testasync] proc restNodeConf(): WakuNodeConf = var conf = defaultWakuNodeConf().valueOr: raiseAssert error - conf.mode = cli_args.WakuMode.Core + conf.entryLayer = EntryLayer.messaging + conf.mode = LogosDeliveryMode.Core conf.listenAddress = parseIpAddress("0.0.0.0") conf.tcpPort = Port(0) conf.discv5UdpPort = Port(0) conf.clusterId = some(3'u16) conf.numShardsInNetwork = 1 - conf.reliabilityEnabled = some(true) conf.rest = true conf.restAddress = parseIpAddress("127.0.0.1") conf.restPort = 0'u16 # bind to an ephemeral port diff --git a/tools/confutils/cli_args.nim b/tools/confutils/cli_args.nim index 1774f1445..1cb6292d2 100644 --- a/tools/confutils/cli_args.nim +++ b/tools/confutils/cli_args.nim @@ -21,6 +21,7 @@ import json import + logos_delivery/api/conf/modes, logos_delivery/waku/factory/[waku_conf, conf_builder/conf_builder, networks_config], logos_delivery/waku/common/[logging], logos_delivery/waku/[ @@ -37,7 +38,7 @@ import ./envvar as confEnvvarDefs, ./envvar_net as confEnvvarNet export confTomlDefs, confTomlNet, confEnvvarDefs, confEnvvarNet, ProtectedShard, - DefaultMaxWakuMessageSizeStr, DefaultAgentString + DefaultMaxWakuMessageSizeStr, DefaultAgentString, modes logScope: topics = "waku cli args" @@ -168,6 +169,20 @@ type WakuNodeConf* = object name: "preset" .}: string + entryLayer* {. + desc: + "Top API layer to run: kernel (transport only), messaging, or channels (messaging + reliable channels).", + defaultValue: EntryLayer.channels, + name: "entry-layer" + .}: EntryLayer + + mode* {. + desc: + "Kernel operating mode: Edge (client-only) or Core (full service node). Applied only for --entry-layer=messaging|channels; ignored for kernel.", + defaultValue: LogosDeliveryMode.Core, + name: "mode" + .}: LogosDeliveryMode + # Opt-typed; desc states the default since the CLI can't auto-show it for Opt.none(). clusterId* {. desc: static(