diff --git a/library/README.md b/library/README.md index e8352c611..e6ddcd4d2 100644 --- a/library/README.md +++ b/library/README.md @@ -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. diff --git a/logos_delivery.nimble b/logos_delivery.nimble index efe118595..8c4d32cb5 100644 --- a/logos_delivery.nimble +++ b/logos_delivery.nimble @@ -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" diff --git a/logos_delivery/api/messaging_client_api.nim b/logos_delivery/api/messaging_client_api.nim index 536ce94fa..5906ab564 100644 --- a/logos_delivery/api/messaging_client_api.nim +++ b/logos_delivery/api/messaging_client_api.nim @@ -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.} diff --git a/logos_delivery/api/reliable_channel_manager_api.nim b/logos_delivery/api/reliable_channel_manager_api.nim index d50b57288..8d3f63ff3 100644 --- a/logos_delivery/api/reliable_channel_manager_api.nim +++ b/logos_delivery/api/reliable_channel_manager_api.nim @@ -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( diff --git a/logos_delivery/channels/api/channel_lifecycle.nim b/logos_delivery/channels/api/channel_lifecycle.nim index b50ba1b63..4f891c61f 100644 --- a/logos_delivery/channels/api/channel_lifecycle.nim +++ b/logos_delivery/channels/api/channel_lifecycle.nim @@ -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, diff --git a/logos_delivery/channels/reliable_channel.nim b/logos_delivery/channels/reliable_channel.nim index 06648a8f7..4daf8a1e2 100644 --- a/logos_delivery/channels/reliable_channel.nim +++ b/logos_delivery/channels/reliable_channel.nim @@ -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, diff --git a/logos_delivery/channels/reliable_channel_manager.nim b/logos_delivery/channels/reliable_channel_manager.nim index e8d579326..8ff176e9a 100644 --- a/logos_delivery/channels/reliable_channel_manager.nim +++ b/logos_delivery/channels/reliable_channel_manager.nim @@ -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`, diff --git a/logos_delivery/logos_delivery.nim b/logos_delivery/logos_delivery.nim index 0b161d6f7..6a6587ab0 100644 --- a/logos_delivery/logos_delivery.nim +++ b/logos_delivery/logos_delivery.nim @@ -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) diff --git a/logos_delivery/messaging/messaging_client.nim b/logos_delivery/messaging/messaging_client.nim index e82c6d6fd..32f5439e5 100644 --- a/logos_delivery/messaging/messaging_client.nim +++ b/logos_delivery/messaging/messaging_client.nim @@ -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. diff --git a/logos_delivery/messaging/messaging_client_lifecycle.nim b/logos_delivery/messaging/messaging_client_lifecycle.nim new file mode 100644 index 000000000..52ea295aa --- /dev/null +++ b/logos_delivery/messaging/messaging_client_lifecycle.nim @@ -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 diff --git a/logos_delivery/waku/factory/networks_config.nim b/logos_delivery/waku/factory/networks_config.nim index 3d1e296fe..339922c02 100644 --- a/logos_delivery/waku/factory/networks_config.nim +++ b/logos_delivery/waku/factory/networks_config.nim @@ -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] = diff --git a/logos_delivery/waku/persistency/backend_thread.nim b/logos_delivery/waku/persistency/backend_thread.nim index e32e5c209..28f328729 100644 --- a/logos_delivery/waku/persistency/backend_thread.nim +++ b/logos_delivery/waku/persistency/backend_thread.nim @@ -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 ─────────────────────────────────────────────────────────── diff --git a/logos_delivery/waku/persistency/persistency.nim b/logos_delivery/waku/persistency/persistency.nim index fca6ad34d..d6ac28a1e 100644 --- a/logos_delivery/waku/persistency/persistency.nim +++ b/logos_delivery/waku/persistency/persistency.nim @@ -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]) diff --git a/logos_delivery/waku/waku_peer_exchange/common.nim b/logos_delivery/waku/waku_peer_exchange/common.nim index 85d7398b1..91f1ac647 100644 --- a/logos_delivery/waku/waku_peer_exchange/common.nim +++ b/logos_delivery/waku/waku_peer_exchange/common.nim @@ -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] diff --git a/logos_delivery/waku/waku_peer_exchange/protocol.nim b/logos_delivery/waku/waku_peer_exchange/protocol.nim index a389938f8..ce1f0cd04 100644 --- a/logos_delivery/waku/waku_peer_exchange/protocol.nim +++ b/logos_delivery/waku/waku_peer_exchange/protocol.nim @@ -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 diff --git a/nimble.lock b/nimble.lock index 18ebde258..03bb7a434 100644 --- a/nimble.lock +++ b/nimble.lock @@ -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": { diff --git a/nix/deps.nix b/nix/deps.nix index 00dea27c2..63b01f8af 100644 --- a/nix/deps.nix +++ b/nix/deps.nix @@ -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"; diff --git a/tests/api/test_node_conf.nim b/tests/api/test_node_conf.nim index df5b18887..2716ffed1 100644 --- a/tests/api/test_node_conf.nim +++ b/tests/api/test_node_conf.nim @@ -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: diff --git a/tests/channels/test_reliable_channel_send_receive.nim b/tests/channels/test_reliable_channel_send_receive.nim index c20f0d0b0..70a8e2702 100644 --- a/tests/channels/test_reliable_channel_send_receive.nim +++ b/tests/channels/test_reliable_channel_send_receive.nim @@ -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]] diff --git a/tests/node/test_wakunode_health_monitor.nim b/tests/node/test_wakunode_health_monitor.nim index 148c93330..c6f9d34e6 100644 --- a/tests/node/test_wakunode_health_monitor.nim +++ b/tests/node/test_wakunode_health_monitor.nim @@ -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 diff --git a/tests/persistency/test_lifecycle.nim b/tests/persistency/test_lifecycle.nim index 2626cb01b..3006672da 100644 --- a/tests/persistency/test_lifecycle.nim +++ b/tests/persistency/test_lifecycle.nim @@ -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"))] diff --git a/tools/confutils/cli_args.nim b/tools/confutils/cli_args.nim index 5228dfff7..f781133af 100644 --- a/tools/confutils/cli_args.nim +++ b/tools/confutils/cli_args.nim @@ -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)