logos-messaging/logos-delivery
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-delivery.git synced 2026-07-22 12:39:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into chore/refresh-merkleproofcache-on-msg-rejection

Browse Source
This commit is contained in:
stubbsta 2026-07-06 10:31:16 +02:00
commit 6b75c6f921
No known key found for this signature in database
22 changed files with 254 additions and 227 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
library/README.md
View File

@ -40,8 +40,17 @@ void *logosdelivery_create_node(
```
Configuration uses flat field names matching `WakuNodeConf` in `tools/confutils/cli_args.nim`.
Use `"preset"` to select a network preset (e.g., `"twn"`, `"logos.dev"`) which auto-configures
entry nodes, cluster ID, sharding, and other network-specific settings.
Use `"preset"` to select a network preset (e.g., `"twn"`, `"logos.dev"`, `"status.prod"`) which
auto-configures entry nodes, cluster ID, sharding, and other network-specific settings.
Available presets:
| Preset | Cluster ID | RLN | Sharding | Network |
| --- | --- | --- | --- | --- |
| `twn` | 1 | on | auto (8 shards) | The Waku Network |
| `logos.dev` | 2 | off | auto (8 shards) | Logos Dev Network |
| `logos.test` | 2 | off | auto (8 shards) | Logos Test Network |
| `status.prod` | 16 | off | auto (1 shard) | Status Production Network |
#### `logosdelivery_start_node`
Starts the node.

2
logos_delivery.nimble
View File

@ -65,7 +65,7 @@ requires "https://github.com/logos-messaging/nim-ffi#v0.1.3"
requires "https://github.com/logos-messaging/nim-sds.git#b12f5ee07c5b764303b51fb948b32a4ade1de3b5"
requires "https://github.com/NagyZoltanPeter/nim-brokers.git#v3.1.1"
requires "https://github.com/NagyZoltanPeter/nim-brokers.git#v3.1.4"
requires "https://github.com/vacp2p/nim-lsquic.git#v0.5.1"
requires "https://github.com/vacp2p/nim-jwt.git#057ec95eb5af0eea9c49bfe9025b3312c95dc5f2"

11
logos_delivery/api/messaging_client_api.nim
View File

@ -1,11 +1,18 @@
import chronos, results
import brokers/[request_broker, broker_context]
import logos_delivery/api/types as api_types
export api_types
export api_types, request_broker, broker_context
# Structural API contract for a messaging client (ops in `messaging/api/*`).
type MessagingApi* = concept c
subscribe(c, contentTopic = ContentTopic) is Future[Result[void, string]]
unsubscribe(c, contentTopic = ContentTopic) is Result[void, string]
send(c, envelope = MessageEnvelope) is Future[Result[RequestId, string]]
# Semi detached MessagingClient send interface.
# Can be used without MessagingClient, under the same context of MessagingClient instance.
RequestBroker:
proc MessagingSend(
envelope: MessageEnvelope
): Future[Result[RequestId, string]] {.async.}

8
logos_delivery/api/reliable_channel_manager_api.nim
View File

@ -5,14 +5,6 @@ import logos_delivery/channels/types as channel_types
export api_types, channel_types
type SendHandler* = proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.
async: (raises: [CatchableError]), gcsafe
.}
## Egress dispatch boundary. Typically wraps `MessagingClient.send`;
## tests inject a fake that records calls and returns canned
## `RequestId`s so the send state machine can be exercised end-to-end
## without a network.
# Structural API contract for the reliable-channel surface (ops in `channels/api/*`).
type ReliableChannelApi* = concept c
createReliableChannel(

12
logos_delivery/channels/api/channel_lifecycle.nim
View File

@ -32,16 +32,7 @@ proc createReliableChannel*(
channelId: ChannelId,
contentTopic: ContentTopic,
senderId: SdsParticipantID,
sendHandler: SendHandler = nil,
): Result[ChannelId, string] =
## Spec entry point. The `sendHandler` and `rng` the channel needs are
## sourced from the owning `ReliableChannelManager` rather than passed
## per call. Encryption is wired up through the `Encrypt`/`Decrypt`
## request brokers — the application installs its own providers
## (or `setNoopEncryption()`) before traffic flows.
##
## `sendHandler` defaults to the manager's default (constructed at mount
## from `MessagingClient.send`); tests pass a fake to bypass the network.
if self.channels.hasKey(channelId):
return err("channel already exists: " & channelId)
@ -60,10 +51,7 @@ proc createReliableChannel*(
epochPeriodSec: DefaultEpochPeriodSec, messagesPerEpoch: DefaultMessagesPerEpoch
)
let effectiveSendHandler = if sendHandler.isNil(): self.sendHandler else: sendHandler
let chn = ReliableChannel.new(
sendHandler = effectiveSendHandler,
channelId = channelId,
contentTopic = contentTopic,
senderId = senderId,

30
logos_delivery/channels/reliable_channel.nim
View File

@ -24,8 +24,9 @@ import libp2p/crypto/crypto as libp2p_crypto
import logos_delivery/api/types
import logos_delivery/api/reliable_channel_manager_api
import logos_delivery/api/events/messaging_client_events
import logos_delivery/api/messaging_client_api
import logos_delivery/api/events/reliable_channel_manager_events
import logos_delivery/messaging/delivery_service/send_service
import logos_delivery/messaging/messaging_client
import logos_delivery/waku/waku_core/topics
import ./segmentation/segmentation
@ -34,7 +35,7 @@ import ./rate_limit_manager/rate_limit_manager
import ./encryption/encryption
export
types, reliable_channel_manager_api, send_service, segmentation, scalable_data_sync,
types, reliable_channel_manager_api, segmentation, scalable_data_sync,
rate_limit_manager, encryption
const LipWireReliableChannelVersion* = "RELIABLE-CHANNEL-API/1"
@ -84,7 +85,6 @@ type
## Spec-defined public type. Fields are private so callers cannot
## mutate internals and break invariants. Getters are added below
## for the few values consumers may need.
sendHandler: SendHandler
channelId: ChannelId
contentTopic: ContentTopic
senderId: SdsParticipantID
@ -263,14 +263,7 @@ proc onReadyToSend(
meta: LipWireReliableChannelVersion.toBytes(),
)
## `sendHandler` is not annotated `(raises: [])`, but this listener is.
## Convert any raise to a Result error so the state machine handles
## both failure modes (Result.err and exception) through one path.
let sendRes =
try:
await self.sendHandler(envelope)
except CatchableError as e:
Result[RequestId, string].err("messaging send raised: " & e.msg)
let sendRes = await MessagingSend.request(self.brokerCtx, envelope)
let messagingReqId = sendRes.valueOr:
MessageErrorEvent.emit(
@ -360,14 +353,9 @@ proc dispatchRepair(self: ReliableChannel, wire: seq[byte]) {.async: (raises: []
meta: LipWireReliableChannelVersion.toBytes(),
)
let sendRes =
try:
await self.sendHandler(envelope)
except CatchableError as e:
Result[RequestId, string].err("messaging send raised: " & e.msg)
if sendRes.isErr():
(await MessagingSend.request(self.brokerCtx, envelope)).isOkOr:
debug "SDS repair rebroadcast dropped: dispatch failed",
channelId = self.channelId, error = sendRes.error
channelId = self.channelId, error = error
proc onMessageReceived(
self: ReliableChannel, messageHash: string, payload: seq[byte]
@ -415,7 +403,6 @@ proc onMessageReceived(
proc new*(
T: type ReliableChannel,
sendHandler: SendHandler,
channelId: ChannelId,
contentTopic: ContentTopic,
senderId: SdsParticipantID,
@ -430,12 +417,7 @@ proc new*(
## should be wiring up. Encryption is delegated to the `Encrypt`/
## `Decrypt` request brokers, so the channel keeps no per-instance
## encryption state either.
##
## `sendHandler` is the egress dispatch. The owning `ReliableChannelManager`
## typically constructs it as a closure over `MessagingClient.send`. Tests
## pass a fake to drive the send state machine without touching the network.
let chn = T(
sendHandler: sendHandler,
channelId: channelId,
contentTopic: contentTopic,
senderId: senderId,

26
logos_delivery/channels/reliable_channel_manager.nim
View File

@ -30,38 +30,14 @@ type
ReliableChannelManager* = ref object ## Implements `ReliableChannelApi`.
channels*: Table[ChannelId, ReliableChannel] ## read by `channels/api.nim`
messagingClient: MessagingClient ## The channel layer chains onto messaging.
sendHandler*: SendHandler
## Default egress dispatch for channels created through this manager.
## Built in `new` as a closure over `MessagingClient.send` so the channel
## layer itself stays callable-only.
brokerCtx*: BrokerContext
proc new*(
T: type ReliableChannelManager,
conf: ReliableChannelManagerConf,
messagingClient: MessagingClient,
brokerCtx: BrokerContext = globalBrokerContext(),
): Result[T, string] =
## The reliable channel layer chains onto the messaging layer: its default
## egress is `MessagingClient.send`, wrapped here so callers never wire the
## handler themselves.
if messagingClient.isNil():
return err("messaging client is required")
let defaultSendHandler: SendHandler = proc(
envelope: MessageEnvelope
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
return await messagingClient.send(envelope)
return ok(
T(
channels: initTable[ChannelId, ReliableChannel](),
messagingClient: messagingClient,
sendHandler: defaultSendHandler,
brokerCtx: brokerCtx,
)
)
return ok(T(channels: initTable[ChannelId, ReliableChannel](), brokerCtx: brokerCtx))
proc start*(self: ReliableChannelManager): Result[void, string] =
## Placeholder: per-channel listeners are installed in `ReliableChannel.new`,

4
logos_delivery/logos_delivery.nim
View File

@ -35,7 +35,7 @@ import logos_delivery/waku/api/events/health_events
export health_events
# Messaging layer
import logos_delivery/messaging/messaging_client
import logos_delivery/messaging/[messaging_client, messaging_client_lifecycle]
export messaging_client
import logos_delivery/messaging/api/[subscription, send]
export subscription, send
@ -103,7 +103,7 @@ proc new*(
return err("failed to create MessagingClient: " & error)
let reliableChannelManager = ReliableChannelManager.new(
layerConf.reliableChannel, messagingClient, waku.brokerCtx
layerConf.reliableChannel, waku.brokerCtx
).valueOr:
return err("failed to create ReliableChannelManager: " & error)

30
logos_delivery/messaging/messaging_client.nim
View File

@ -2,6 +2,7 @@
## lifecycle. The public operations (subscribe / unsubscribe / send) live in
## `messaging/api.nim`.
import results, chronos
import chronicles
import
logos_delivery/api/messaging_client_api,
logos_delivery/waku/waku,
@ -17,11 +18,12 @@ type
## follow-up PR. Today it only carries the p2p reliability toggle.
useP2PReliability*: bool
MessagingClient* = ref object ## Implements `MessagingApi`.
MessagingClient* = ref object
brokerCtx*: BrokerContext
waku*: Waku ## The Waku kernel this layer drives; read by `messaging/api/*`.
sendService*: SendService
recvService*: RecvService
started: bool
started*: bool
proc new*(
T: type MessagingClient, conf: MessagingClientConf, waku: Waku
@ -30,22 +32,14 @@ proc new*(
## for transport while exposing its own send/recv API.
let sendService = ?SendService.new(conf.useP2PReliability, waku)
let recvService = RecvService.new(waku)
return ok(T(waku: waku, sendService: sendService, recvService: recvService))
proc start*(self: MessagingClient): Result[void, string] =
if self.started:
return ok()
self.recvService.startRecvService()
self.sendService.startSendService()
self.started = true
ok()
proc stop*(self: MessagingClient) {.async.} =
if not self.started:
return
await self.sendService.stopSendService()
await self.recvService.stopRecvService()
self.started = false
return ok(
T(
waku: waku,
sendService: sendService,
recvService: recvService,
brokerCtx: waku.brokerCtx,
)
)
proc checkApiAvailability*(self: MessagingClient): Result[void, string] =
## Shared guard for the api operation module.

37
logos_delivery/messaging/messaging_client_lifecycle.nim Normal file
View File

@ -0,0 +1,37 @@
## Messaging layer core: the `MessagingClient` type plus its construction and
## lifecycle. The public operations (subscribe / unsubscribe / send) live in
## `messaging/api.nim`.
import results, chronos
import chronicles
import
logos_delivery/messaging/messaging_client,
logos_delivery/messaging/api/send,
logos_delivery/api/messaging_client_api,
logos_delivery/waku/waku,
logos_delivery/messaging/delivery_service/[recv_service, send_service]
# Surfaces the messaging API interface (and its Message* events) to consumers.
export messaging_client
proc start*(self: MessagingClient): Result[void, string] =
if self.started:
return ok()
self.recvService.startRecvService()
self.sendService.startSendService()
?MessagingSend.setProvider(
self.brokerCtx,
proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
return await self.send(envelope),
)
self.started = true
ok()
proc stop*(self: MessagingClient) {.async.} =
if not self.started:
return
MessagingSend.clearProvider(self.brokerCtx)
await self.sendService.stopSendService()
await self.recvService.stopRecvService()
self.started = false

33
logos_delivery/waku/factory/networks_config.nim
View File

@ -123,6 +123,39 @@ proc LogosTestConf*(T: type NetworkPresetConf): NetworkPresetConf =
],
)
# cluster-id=16 (Status Production Network)
# Cluster configuration for the `status.prod` network that Status runs on.
# RLN is disabled. Starting from the logos-delivery integration, status.prod
# defaults to auto-sharding with a single shard (numShardsInCluster = 1).
# Bootstrap is done through the status.prod DNS discovery enrtree plus the
# fleet boot nodes.
# Source: https://fleets.waku.org/ and each host's `/config.toml`.
proc StatusProdConf*(T: type NetworkPresetConf): NetworkPresetConf =
const ZeroChainId = 0'u256
return NetworkPresetConf(
maxMessageSize: "1024KiB",
clusterId: 16,
rlnRelay: false,
rlnRelayEthContractAddress: "",
rlnRelayDynamic: false,
rlnRelayChainId: ZeroChainId,
rlnEpochSizeSec: 0,
rlnRelayUserMessageLimit: 0,
shardingConf: ShardingConf(kind: AutoSharding, numShardsInCluster: 1),
enableKadDiscovery: false,
kadBootstrapNodes: @[],
mix: false,
p2pReliability: false,
discv5Discovery: true,
discv5BootstrapNodes: @[],
entryNodes: @[
"enrtree://AMOJVZX4V6EXP7NTJPMAYJYST2QP6AJXYW76IU6VGJS7UVSNDYZG4@boot.prod.status.nodes.status.im",
"/dns4/boot-01.do-ams3.status.prod.status.im/tcp/30303/p2p/16Uiu2HAmAR24Mbb6VuzoyUiGx42UenDkshENVDj4qnmmbabLvo31",
"/dns4/boot-01.gc-us-central1-a.status.prod.status.im/tcp/30303/p2p/16Uiu2HAm8mUZ18tBWPXDQsaF7PbCKYA35z7WB2xNZH2EVq1qS8LJ",
"/dns4/boot-01.ac-cn-hongkong-c.status.prod.status.im/tcp/30303/p2p/16Uiu2HAmGwcE8v7gmJNEWFtZtojYpPMTHy2jBLL6xRk33qgDxFWX",
],
)
proc validateShards*(
shardingConf: ShardingConf, shards: seq[uint16]
): Result[void, string] =

6
logos_delivery/waku/persistency/backend_thread.nim
View File

@ -162,13 +162,13 @@ proc registerProviders(backend: KvBackend, ctx: BrokerContext): Result[void, str
return ok()
proc clearProviders(ctx: BrokerContext) =
proc clearProviders(ctx: BrokerContext) {.async.} =
KvGet.clearProvider(ctx)
KvExists.clearProvider(ctx)
KvScan.clearProvider(ctx)
KvCount.clearProvider(ctx)
KvDelete.clearProvider(ctx)
PersistEvent.dropAllListeners(ctx)
await PersistEvent.dropAllListeners(ctx)
# ── thread proc ─────────────────────────────────────────────────────────
@ -217,7 +217,7 @@ proc storageThreadMain(arg: ptr StorageThreadArg) {.thread.} =
except CatchableError as e:
error "storage thread loop crashed", err = e.msg
clearProviders(arg.ctx)
waitFor clearProviders(arg.ctx)
backend.close()
# ── lifecycle ───────────────────────────────────────────────────────────

2
logos_delivery/waku/persistency/persistency.nim
View File

@ -272,7 +272,7 @@ proc hasJob*(p: Persistency, jobId: string): bool {.inline.} =
proc persist*(t: Job, ops: seq[TxOp]): Future[void] {.async.} =
## Emit a batched persist event. The handler treats >1 ops as a single
## BEGIN IMMEDIATE/COMMIT transaction (see backend_sqlite.applyOps).
await PersistEvent.emit(t.context, PersistEvent(ops: ops))
PersistEvent.emit(t.context, PersistEvent(ops: ops))
proc persist*(t: Job, op: TxOp): Future[void] {.async.} =
await persist(t, @[op])

1
logos_delivery/waku/waku_peer_exchange/common.nim
View File

@ -17,5 +17,6 @@ const
decodeRpcFailure* = "decode_rpc_failure"
retrievePeersDiscv5Error* = "retrieve_peers_discv5_failure"
pxFailure* = "px_failure"
streamClosedFailure* = "stream_closed_failure"
type WakuPeerExchangeResult*[T] = Result[T, PeerExchangeResponseStatus]

81
logos_delivery/waku/waku_peer_exchange/protocol.nim
View File

@ -6,6 +6,7 @@ import
chronos,
metrics,
libp2p/protocols/protocol,
libp2p/stream/connection,
libp2p/crypto/crypto,
eth/p2p/discoveryv5/enr
import
@ -38,18 +39,20 @@ type WakuPeerExchange* = ref object of LPProtocol
proc respond(
wpx: WakuPeerExchange, enrs: seq[enr.Record], conn: Connection
): Future[WakuPeerExchangeResult[void]] {.async, gcsafe.} =
): Future[WakuPeerExchangeResult[void]] {.async: (raises: [CancelledError]), gcsafe.} =
let rpc = PeerExchangeRpc.makeResponse(enrs.mapIt(PeerExchangePeerInfo(enr: it.raw)))
try:
await conn.writeLP(rpc.encode().buffer)
except CatchableError as exc:
error "exception when trying to send a respond", error = getCurrentExceptionMsg()
waku_px_errors.inc(labelValues = [exc.msg])
except LPStreamError as exc:
# Remote closed the stream before we responded - expected during peer churn.
debug "peer exchange response not delivered: stream closed",
peerId = conn.peerId, error = exc.msg
waku_px_errors.inc(labelValues = [streamClosedFailure])
return err(
(
status_code: PeerExchangeResponseStatusCode.DIAL_FAILURE,
status_desc: some("exception dialing peer: " & exc.msg),
status_desc: some("stream closed before response: " & exc.msg),
)
)
@ -60,18 +63,20 @@ proc respondError(
status_code: PeerExchangeResponseStatusCode,
status_desc: Option[string],
conn: Connection,
): Future[WakuPeerExchangeResult[void]] {.async, gcsafe.} =
): Future[WakuPeerExchangeResult[void]] {.async: (raises: [CancelledError]), gcsafe.} =
let rpc = PeerExchangeRpc.makeErrorResponse(status_code, status_desc)
try:
await conn.writeLP(rpc.encode().buffer)
except CatchableError as exc:
error "exception when trying to send a respond", error = getCurrentExceptionMsg()
waku_px_errors.inc(labelValues = [exc.msg])
except LPStreamError as exc:
# Remote closed the stream before we responded - expected during peer churn.
debug "peer exchange error response not delivered: stream closed",
peerId = conn.peerId, error = exc.msg
waku_px_errors.inc(labelValues = [streamClosedFailure])
return err(
(
status_code: PeerExchangeResponseStatusCode.SERVICE_UNAVAILABLE,
status_desc: some("exception dialing peer: " & exc.msg),
status_desc: some("stream closed before response: " & exc.msg),
)
)
@ -128,20 +133,12 @@ proc initProtocolHandler(wpx: WakuPeerExchange) =
wpx.requestRateLimiter.checkUsageLimit(WakuPeerExchangeCodec, conn):
try:
buffer = await conn.readLp(DefaultMaxRpcSize.int)
except CatchableError as exc:
error "exception when handling px request", error = getCurrentExceptionMsg()
waku_px_errors.inc(labelValues = [exc.msg])
(
try:
await wpx.respondError(
PeerExchangeResponseStatusCode.BAD_REQUEST, some(exc.msg), conn
)
except CatchableError:
error "could not send error response", error = getCurrentExceptionMsg()
return
).isOkOr:
error "Failed to respond with BAD_REQUEST:", error = $error
except LPStreamError as exc:
# Remote disconnected before sending a full request - expected churn;
# no point responding on a dead stream.
debug "peer exchange request not received: stream closed",
peerId = conn.peerId, error = exc.msg
waku_px_errors.inc(labelValues = [streamClosedFailure])
return
let decBuf = PeerExchangeRpc.decode(buffer).valueOr:
@ -149,42 +146,30 @@ proc initProtocolHandler(wpx: WakuPeerExchange) =
error "Failed to decode PeerExchange request", error = $error
(
try:
await wpx.respondError(
PeerExchangeResponseStatusCode.BAD_REQUEST, some($error), conn
)
except CatchableError:
error "could not send error response decode",
error = getCurrentExceptionMsg()
return
await wpx.respondError(
PeerExchangeResponseStatusCode.BAD_REQUEST, some($error), conn
)
).isOkOr:
error "Failed to respond with BAD_REQUEST:", error = $error
error "Failed to respond with BAD_REQUEST", error = $error
return
let enrs = wpx.getEnrsFromStore(decBuf.request.numPeers)
info "peer exchange request received"
trace "px enrs to respond", enrs = $enrs
try:
(await wpx.respond(enrs, conn)).isErrOr:
waku_px_peers_sent.inc(enrs.len().int64())
except CatchableError:
error "could not send response", error = getCurrentExceptionMsg()
(await wpx.respond(enrs, conn)).isErrOr:
waku_px_peers_sent.inc(enrs.len().int64())
do:
defer:
# close, no data is expected
await conn.closeWithEof()
try:
(
await wpx.respondError(
PeerExchangeResponseStatusCode.TOO_MANY_REQUESTS, none(string), conn
)
).isOkOr:
error "Failed to respond with TOO_MANY_REQUESTS:", error = $error
except CatchableError:
error "could not send error response", error = getCurrentExceptionMsg()
return
(
await wpx.respondError(
PeerExchangeResponseStatusCode.TOO_MANY_REQUESTS, none(string), conn
)
).isOkOr:
error "Failed to respond with TOO_MANY_REQUESTS", error = $error
wpx.handler = handler
wpx.codec = WakuPeerExchangeCodec

6
nimble.lock
View File

@ -328,8 +328,8 @@
}
},
"brokers": {
"version": "#v3.1.1",
"vcsRevision": "a7316a35f1b62e3497ae8ee0fc1aace74df0beb2",
"version": "#v3.1.4",
"vcsRevision": "8ae9e963b0b4478c93e6f888be6a46654da787de",
"url": "https://github.com/NagyZoltanPeter/nim-brokers.git",
"downloadMethod": "git",
"dependencies": [
@ -341,7 +341,7 @@
"cbor_serialization"
],
"checksums": {
"sha1": "4447d7c1f9da14ae439afb23aee45116ce2ecb40"
"sha1": "2949398033c1b3a2586f0afd7e92f46b9a0a412e"
}
},
"stint": {

5
nix/deps.nix
View File

@ -159,8 +159,8 @@
brokers = pkgs.fetchgit {
url = "https://github.com/NagyZoltanPeter/nim-brokers.git";
rev = "a7316a35f1b62e3497ae8ee0fc1aace74df0beb2";
sha256 = "1990270n88jm0i48g07zr4vq2nn79g7gymf28f3g5ak42g33l7rm";
rev = "8ae9e963b0b4478c93e6f888be6a46654da787de";
sha256 = "0r3vrc6h9jiwhd3p4sy3j8gx3k5174s77qcp2wrswclvfvbskw5y";
fetchSubmodules = true;
};
@ -277,7 +277,6 @@
};
sds = pkgs.fetchgit {
# Keep in sync with the nim-sds pin in nimble.lock.
url = "https://github.com/logos-messaging/nim-sds.git";
rev = "b12f5ee07c5b764303b51fb948b32a4ade1de3b5";
sha256 = "1z8f0v1ww7y6zssdacjxfs6s4862dwckw25df3yn1v0qnz40rpc8";

19
tests/api/test_node_conf.nim
View File

@ -241,6 +241,25 @@ suite "WakuNodeConf - preset integration":
check:
wakuConf.clusterId == 2
test "StatusProd preset applies StatusProdConf":
## Given
var conf = defaultWakuNodeConf().valueOr:
raiseAssert error
conf.preset = "status.prod"
## When
let wakuConfRes = conf.toWakuConf()
## Then
require wakuConfRes.isOk()
let wakuConf = wakuConfRes.get()
require wakuConf.validate().isOk()
check:
wakuConf.clusterId == 16
wakuConf.shardingConf.kind == AutoSharding
wakuConf.shardingConf.numShardsInCluster == 1
wakuConf.rlnRelayConf.isNone()
test "Invalid preset returns error":
## Given
var conf = defaultWakuNodeConf().valueOr:

140
tests/channels/test_reliable_channel_send_receive.nim
View File

@ -11,6 +11,7 @@ import logos_delivery
import logos_delivery/waku/[waku_node, waku_core]
import logos_delivery/waku/factory/waku_conf
import logos_delivery/api/events/messaging_client_events as waku_message_events
import logos_delivery/api/messaging_client_api
import tools/confutils/cli_args
import logos_delivery/channels/reliable_channel_manager
@ -187,16 +188,16 @@ suite "Reliable Channel - send state machine":
setNoopEncryption()
var sendCalls = 0
let fakeSend: SendHandler = proc(
env: MessageEnvelope
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
sendCalls.inc
return ok(fakeMsgReqId)
MessagingSend.replaceProvider(
brokerCtx,
proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
sendCalls.inc
return ok(fakeMsgReqId),
).isOkOr:
raiseAssert "replaceProvider failed: " & error
discard manager
.createReliableChannel(
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
)
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
.expect("createReliableChannel")
let sentFut = newFuture[RequestId]("channel-sent")
@ -252,17 +253,17 @@ suite "Reliable Channel - send state machine":
setNoopEncryption()
var msgReqIds: seq[RequestId]
let fakeSend: SendHandler = proc(
env: MessageEnvelope
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
let id = RequestId("fake-msg-req-" & $(msgReqIds.len + 1))
msgReqIds.add(id)
return ok(id)
MessagingSend.replaceProvider(
brokerCtx,
proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
let id = RequestId("fake-msg-req-" & $(msgReqIds.len + 1))
msgReqIds.add(id)
return ok(id),
).isOkOr:
raiseAssert "replaceProvider failed: " & error
discard manager
.createReliableChannel(
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
)
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
.expect("createReliableChannel")
let sentFut = newFuture[RequestId]("channel-sent")
@ -354,28 +355,27 @@ suite "Reliable Channel - send state machine":
var msgReqIds: seq[RequestId]
var sendsReturned = 0
let fakeSend: SendHandler = proc(
env: MessageEnvelope
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
## Call 2 fires the first segment's terminal event and then
## yields, so the listener task runs while the second segment
## is still mid-`await` in `onReadyToSend` — the exact race
## window the regression test targets.
let id = RequestId("race-msg-req-" & $(msgReqIds.len + 1))
msgReqIds.add(id)
if msgReqIds.len == 2:
waku_message_events.MessageSentEvent.emit(
brokerCtx,
waku_message_events.MessageSentEvent(requestId: msgReqIds[0], messageHash: ""),
)
await sleepAsync(50.milliseconds)
sendsReturned.inc()
return ok(id)
MessagingSend.replaceProvider(
brokerCtx,
proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
## Call 2 fires the first segment's terminal event and then
## yields, so the listener task runs while the second segment
## is still mid-`await` in `onReadyToSend` — the exact race
## window the regression test targets.
let id = RequestId("race-msg-req-" & $(msgReqIds.len + 1))
msgReqIds.add(id)
if msgReqIds.len == 2:
MessageSentEvent.emit(
brokerCtx, MessageSentEvent(requestId: msgReqIds[0], messageHash: "")
)
await sleepAsync(50.milliseconds)
sendsReturned.inc()
return ok(id),
).isOkOr:
raiseAssert "replaceProvider failed: " & error
discard manager
.createReliableChannel(
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
)
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
.expect("createReliableChannel")
var finalisedReqIds: seq[RequestId]
@ -457,15 +457,15 @@ suite "Reliable Channel - SDS persistence":
setNoopEncryption()
let fakeSend: SendHandler = proc(
env: MessageEnvelope
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
return ok(RequestId("persist-msg-req-1"))
MessagingSend.replaceProvider(
globalBrokerContext(),
proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
return ok(RequestId("persist-msg-req-1")),
).isOkOr:
raiseAssert "replaceProvider failed: " & error
discard manager
.createReliableChannel(
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
)
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
.expect("createReliableChannel")
discard (await manager.send(channelId, "persist me".toBytes())).expect("send")
@ -798,17 +798,17 @@ suite "Reliable Channel - SDS protocol semantics":
setNoopEncryption()
var capturedWires: seq[seq[byte]]
let fakeSend: SendHandler = proc(
env: MessageEnvelope
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
## Noop encryption is identity, so the envelope payload IS the SDS wire.
capturedWires.add(env.payload)
return ok(RequestId("semantics-req-" & $capturedWires.len))
MessagingSend.replaceProvider(
brokerCtx,
proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
## Noop encryption is identity, so the envelope payload IS the SDS wire.
capturedWires.add(envelope.payload)
return ok(RequestId("semantics-req-" & $capturedWires.len)),
).isOkOr:
raiseAssert "replaceProvider failed: " & error
discard manager
.createReliableChannel(
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
)
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
.expect("createReliableChannel")
let remotePeer =
@ -866,16 +866,17 @@ suite "Reliable Channel - SDS protocol semantics":
setNoopEncryption()
var capturedWires: seq[seq[byte]]
let fakeSend: SendHandler = proc(
env: MessageEnvelope
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
capturedWires.add(env.payload)
return ok(RequestId("ack-req-" & $capturedWires.len))
MessagingSend.replaceProvider(
brokerCtx,
proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
## Noop encryption is identity, so the envelope payload IS the SDS wire.
capturedWires.add(envelope.payload)
return ok(RequestId("ack-req-" & $capturedWires.len)),
).isOkOr:
raiseAssert "replaceProvider failed: " & error
discard manager
.createReliableChannel(
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
)
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
.expect("createReliableChannel")
discard (await manager.send(channelId, "needs ack".toBytes())).expect("send")
@ -1102,16 +1103,17 @@ suite "Reliable Channel - SDS protocol semantics":
setNoopEncryption()
var capturedWires: seq[seq[byte]]
let fakeSend: SendHandler = proc(
env: MessageEnvelope
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
capturedWires.add(env.payload)
return ok(RequestId("unique-req-" & $capturedWires.len))
MessagingSend.replaceProvider(
brokerCtx,
proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
## Noop encryption is identity, so the envelope payload IS the SDS wire.
capturedWires.add(envelope.payload)
return ok(RequestId("unique-req-" & $capturedWires.len)),
).isOkOr:
raiseAssert "replaceProvider failed: " & error
discard manager
.createReliableChannel(
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
)
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
.expect("createReliableChannel")
var deliveries: seq[seq[byte]]

1
tests/node/test_wakunode_health_monitor.nim
View File

@ -29,6 +29,7 @@ import logos_delivery/waku/node/subscription_manager
import logos_delivery/waku/waku
import logos_delivery/waku/factory/waku_state_info
import logos_delivery/messaging/messaging_client
import logos_delivery/messaging/messaging_client_lifecycle
const MockDLow = 4 # Mocked GossipSub DLow value

10
tests/persistency/test_lifecycle.nim
View File

@ -156,7 +156,7 @@ suite "Persistency lifecycle":
let ev = PersistEvent(
ops: @[TxOp(category: "msg", key: k, kind: txPut, payload: payloadBytes("hello"))]
)
await PersistEvent.emit(t.context, ev)
PersistEvent.emit(t.context, ev)
let ckOk2 = await t.pollExists("msg", k)
check ckOk2
@ -178,7 +178,7 @@ suite "Persistency lifecycle":
check a.context != b.context
let k = key("shared", 1'i64)
await PersistEvent.emit(
PersistEvent.emit(
a.context,
PersistEvent(
ops: @[
@ -188,7 +188,7 @@ suite "Persistency lifecycle":
]
),
)
await PersistEvent.emit(
PersistEvent.emit(
b.context,
PersistEvent(
ops: @[
@ -255,7 +255,7 @@ suite "Persistency lifecycle":
ops.add(
TxOp(category: "msg", key: key("c", i), kind: txPut, payload: payloadBytes($i))
)
await PersistEvent.emit(t.context, PersistEvent(ops: ops))
PersistEvent.emit(t.context, PersistEvent(ops: ops))
# Wait for the last insert to land.
let ckOk5 = await t.pollExists("msg", key("c", 5'i64))
check ckOk5
@ -285,7 +285,7 @@ suite "Persistency lifecycle":
let r1 = aw7.get()
check r1.existed == false
await PersistEvent.emit(
PersistEvent.emit(
t.context,
PersistEvent(
ops: @[TxOp(category: "msg", key: k, kind: txPut, payload: payloadBytes("v"))]

4
tools/confutils/cli_args.nim
View File

@ -175,7 +175,7 @@ type WakuNodeConf* = object
preset* {.
desc:
"Network preset to use. 'twn' is The RLN-protected Waku Network (cluster 1). 'logos.dev' is the Logos Dev Network (cluster 2). 'logos.test' is the Logos Test Network (cluster 2). Overrides other values.",
"Network preset to use. 'twn' is The RLN-protected Waku Network (cluster 1). 'logos.dev' is the Logos Dev Network (cluster 2). 'logos.test' is the Logos Test Network (cluster 2). 'status.prod' is the Status Production Network (cluster 16, RLN off, auto-sharding with 1 shard). Overrides other values.",
defaultValue: "",
name: "preset"
.}: string
@ -1001,6 +1001,8 @@ proc toNetworkPresetConf(
ok(some(NetworkPresetConf.LogosDevConf()))
of "logos.test", "logostest":
ok(some(NetworkPresetConf.LogosTestConf()))
of "status.prod", "statusprod":
ok(some(NetworkPresetConf.StatusProdConf()))
else:
err("Invalid --preset value passed: " & lcPreset)