mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-26 06:23:14 +00:00
avoid sendHandler with concepts
This commit is contained in:
Ivan FB
parent
440bd01cd8
commit
1ffee9600c
@ -4,8 +4,17 @@ import logos_delivery/api/types as api_types
|
||||
|
||||
export api_types
|
||||
|
||||
type MessagingSender* = concept c
|
||||
## The narrow egress capability the reliable-channel layer depends on: just
|
||||
## `send`. A layer above messaging depends on this capability, not on the
|
||||
## concrete `MessagingClient`. Dot-call form (`c.send(...)`) so it is
|
||||
## satisfied by both a real `send` proc (the production `MessagingClient`)
|
||||
## and a `send` proc-typed field (a test double).
|
||||
c.send(MessageEnvelope) is Future[Result[RequestId, string]]
|
||||
|
||||
# Structural API contract for a messaging client (ops in `messaging/api/*`).
|
||||
# Refines `MessagingSender`, so the send capability is stated once.
|
||||
type MessagingApi* = concept c
|
||||
c is MessagingSender
|
||||
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]]
|
||||
|
||||
@ -1,19 +1,17 @@
|
||||
import chronos, results
|
||||
|
||||
import logos_delivery/api/types as api_types
|
||||
import logos_delivery/api/messaging_client_api
|
||||
import logos_delivery/channels/types as channel_types
|
||||
|
||||
export api_types, channel_types
|
||||
# `messaging_client_api` is re-exported for `MessagingSender`, the egress
|
||||
# capability the generic `ReliableChannel[M]`/`ReliableChannelManager[M]` need.
|
||||
export api_types, messaging_client_api, 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/*`).
|
||||
# Structural API contract for the reliable-channel surface. This is the
|
||||
# node-bound consumer view (the messaging node is already bound), so it is
|
||||
# satisfied by `LogosDelivery` — the concentrator that owns the node — rather
|
||||
# than by the manager, whose `createReliableChannel` takes the node explicitly.
|
||||
type ReliableChannelApi* = concept c
|
||||
createReliableChannel(
|
||||
c, channelId = ChannelId, contentTopic = ContentTopic, senderId = SdsParticipantID
|
||||
|
||||
@ -8,7 +8,7 @@ import logos_delivery/channels/reliable_channel_manager
|
||||
import logos_delivery/channels/reliable_channel
|
||||
import logos_delivery/waku/persistency/sds_persistency
|
||||
|
||||
# ReliableChannel, SendHandler, config and wire-version markers.
|
||||
# ReliableChannel, config and wire-version markers.
|
||||
export reliable_channel
|
||||
|
||||
const SdsJobId = "sds"
|
||||
@ -32,16 +32,9 @@ 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.
|
||||
## Spec entry point. The messaging node the channel dispatches through is the
|
||||
## one the manager owns (`self.messaging`), so callers address channels by id.
|
||||
if self.channels.hasKey(channelId):
|
||||
return err("channel already exists: " & channelId)
|
||||
|
||||
@ -60,10 +53,8 @@ proc createReliableChannel*(
|
||||
epochPeriodSec: DefaultEpochPeriodSec, messagesPerEpoch: DefaultMessagesPerEpoch
|
||||
)
|
||||
|
||||
let effectiveSendHandler = if sendHandler.isNil(): self.sendHandler else: sendHandler
|
||||
|
||||
let chn = ReliableChannel.new(
|
||||
sendHandler = effectiveSendHandler,
|
||||
messaging = self.messaging,
|
||||
channelId = channelId,
|
||||
contentTopic = contentTopic,
|
||||
senderId = senderId,
|
||||
|
||||
@ -21,11 +21,11 @@ import bearssl/rand
|
||||
import stew/byteutils
|
||||
import libp2p/crypto/crypto as libp2p_crypto
|
||||
|
||||
import brokers/broker_context
|
||||
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/events/reliable_channel_manager_events
|
||||
import logos_delivery/messaging/delivery_service/send_service
|
||||
import logos_delivery/waku/waku_core/topics
|
||||
|
||||
import ./segmentation/segmentation
|
||||
@ -34,7 +34,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"
|
||||
@ -80,11 +80,15 @@ type
|
||||
## and that scan is correct precisely because the outer iteration
|
||||
## order matches the order `send` pushed entries.
|
||||
|
||||
ReliableChannel* = ref object
|
||||
ReliableChannel*[M: MessagingSender] = ref object
|
||||
## 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
|
||||
##
|
||||
## Generic over `M`, the messaging egress: the channel calls `M.send`
|
||||
## directly (compile-time dispatch). Production
|
||||
## instantiates `M = MessagingClient`; tests instantiate a one-proc double.
|
||||
messaging: M
|
||||
channelId: ChannelId
|
||||
contentTopic: ContentTopic
|
||||
senderId: SdsParticipantID
|
||||
@ -195,7 +199,7 @@ proc onMessageFinal(
|
||||
proc markSegmentFailed(self: ReliableChannel, channelReqId: RequestId) =
|
||||
try:
|
||||
self.channelReqs[channelReqId].failedCount.inc()
|
||||
except KeyError as e:
|
||||
except system.KeyError as e:
|
||||
error "unreachable: channelReqId not found in markSegmentFailed",
|
||||
channelReqId = $channelReqId, error = e.msg
|
||||
return
|
||||
@ -206,7 +210,7 @@ proc markSegmentInflight(
|
||||
) =
|
||||
try:
|
||||
self.channelReqs[channelReqId].inflightMessagingIds.add(messagingReqId)
|
||||
except KeyError as e:
|
||||
except system.KeyError as e:
|
||||
error "unreachable: channelReqId not found in markSegmentInflight",
|
||||
channelReqId = $channelReqId, error = e.msg
|
||||
|
||||
@ -263,12 +267,12 @@ 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.
|
||||
## `M.send` (e.g. `MessagingClient.send`) is not annotated `(raises: [])`,
|
||||
## but this listener is. Convert any raise to a Result error so the state
|
||||
## machine handles both failure modes through one path.
|
||||
let sendRes =
|
||||
try:
|
||||
await self.sendHandler(envelope)
|
||||
await self.messaging.send(envelope)
|
||||
except CatchableError as e:
|
||||
Result[RequestId, string].err("messaging send raised: " & e.msg)
|
||||
|
||||
@ -360,14 +364,16 @@ proc dispatchRepair(self: ReliableChannel, wire: seq[byte]) {.async: (raises: []
|
||||
meta: LipWireReliableChannelVersion.toBytes(),
|
||||
)
|
||||
|
||||
## `M.send` may raise; this proc is `(raises: [])`, so funnel any raise into
|
||||
## a Result the same way the main dispatch path does.
|
||||
let sendRes =
|
||||
try:
|
||||
await self.sendHandler(envelope)
|
||||
await self.messaging.send(envelope)
|
||||
except CatchableError as e:
|
||||
Result[RequestId, string].err("messaging send raised: " & e.msg)
|
||||
if sendRes.isErr():
|
||||
sendRes.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]
|
||||
@ -413,9 +419,9 @@ proc onMessageReceived(
|
||||
for content in deliverable:
|
||||
self.reportReceived(content)
|
||||
|
||||
proc new*(
|
||||
proc new*[M: MessagingSender](
|
||||
T: type ReliableChannel,
|
||||
sendHandler: SendHandler,
|
||||
messaging: M,
|
||||
channelId: ChannelId,
|
||||
contentTopic: ContentTopic,
|
||||
senderId: SdsParticipantID,
|
||||
@ -423,19 +429,15 @@ proc new*(
|
||||
sdsConfig: SdsConfig,
|
||||
rateConfig: RateLimitConfig,
|
||||
brokerCtx: BrokerContext = globalBrokerContext(),
|
||||
): T =
|
||||
): ReliableChannel[M] =
|
||||
## Pipeline handlers (segmentation/SDS/rate-limit) are constructed
|
||||
## inside the channel rather than handed in by the caller — they are
|
||||
## implementation details of the channel, not knobs the API consumer
|
||||
## 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,
|
||||
let chn = ReliableChannel[M](
|
||||
messaging: messaging,
|
||||
channelId: channelId,
|
||||
contentTopic: contentTopic,
|
||||
senderId: senderId,
|
||||
|
||||
@ -13,8 +13,6 @@ import stew/byteutils
|
||||
|
||||
import brokers/broker_context
|
||||
|
||||
import logos_delivery/messaging/messaging_client
|
||||
import logos_delivery/messaging/api/send
|
||||
import logos_delivery/api/types
|
||||
import logos_delivery/api/reliable_channel_manager_api
|
||||
|
||||
@ -28,37 +26,25 @@ type
|
||||
## channel API. Placeholder for now (segmentation / SDS / rate-limit defaults
|
||||
## will move here in a follow-up PR); kept so each layer owns its own config.
|
||||
|
||||
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.
|
||||
ReliableChannelManager*[M: MessagingSender] = ref object
|
||||
## Owns the set of channels and the messaging node they dispatch through.
|
||||
## Generic over the egress `M`; hands the node to every channel it creates.
|
||||
## Holding the node here keeps `createReliableChannel` node-free — the
|
||||
## node-bound consumer surface the `ReliableChannelApi` concept describes.
|
||||
channels*: Table[ChannelId, ReliableChannel[M]] ## read by `channels/api.nim`
|
||||
messaging*: M ## egress node, passed on to each `ReliableChannel`
|
||||
brokerCtx*: BrokerContext
|
||||
|
||||
proc new*(
|
||||
proc new*[M: MessagingSender](
|
||||
T: type ReliableChannelManager,
|
||||
conf: ReliableChannelManagerConf,
|
||||
messagingClient: MessagingClient,
|
||||
messaging: M,
|
||||
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)
|
||||
|
||||
): Result[ReliableChannelManager[M], string] =
|
||||
return ok(
|
||||
T(
|
||||
channels: initTable[ChannelId, ReliableChannel](),
|
||||
messagingClient: messagingClient,
|
||||
sendHandler: defaultSendHandler,
|
||||
ReliableChannelManager[M](
|
||||
channels: initTable[ChannelId, ReliableChannel[M]](),
|
||||
messaging: messaging,
|
||||
brokerCtx: brokerCtx,
|
||||
)
|
||||
)
|
||||
|
||||
@ -74,7 +74,7 @@ type
|
||||
LogosDelivery* = ref object ## Entry point. Holds one instance of each API layer.
|
||||
waku*: Waku
|
||||
messagingClient*: MessagingClient
|
||||
reliableChannelManager*: ReliableChannelManager
|
||||
reliableChannelManager*: ReliableChannelManager[MessagingClient]
|
||||
|
||||
proc init*(T: type LogosDeliveryConf, wakuConf: WakuConf): LogosDeliveryConf =
|
||||
## Builds the aggregated config from a `WakuConf`. The messaging / reliable
|
||||
@ -154,5 +154,5 @@ proc isOnline*(self: LogosDelivery): Future[Result[bool, string]] {.async.} =
|
||||
static:
|
||||
doAssert Waku is KernelApi
|
||||
doAssert MessagingClient is MessagingApi
|
||||
doAssert ReliableChannelManager is ReliableChannelApi
|
||||
doAssert ReliableChannelManager[MessagingClient] is ReliableChannelApi
|
||||
doAssert LogosDelivery is LogosDeliveryApi
|
||||
|
||||
@ -38,6 +38,27 @@ proc createApiNodeConf(): WakuNodeConf =
|
||||
conf.rest = false
|
||||
return conf
|
||||
|
||||
type MessagingClientTest = ref object
|
||||
## Test double for the messaging node used by the send-side tests. `send` is
|
||||
## a proc-typed field, so it satisfies the (dot-form) `MessagingSender`
|
||||
## concept directly; the manager owns it as `manager.messaging`, and each
|
||||
## test scripts the egress with `manager.messaging.send = ...`.
|
||||
send: proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.
|
||||
async: (raises: [CatchableError]), gcsafe
|
||||
.}
|
||||
|
||||
static:
|
||||
doAssert MessagingClientTest is MessagingSender
|
||||
|
||||
proc newTestManager(
|
||||
brokerCtx: BrokerContext
|
||||
): ReliableChannelManager[MessagingClientTest] =
|
||||
## Manager over the send double. `manager.messaging` is the double, so tests
|
||||
## script the egress with `manager.messaging.send = ...`.
|
||||
ReliableChannelManager
|
||||
.new(ReliableChannelManagerConf(), MessagingClientTest(), brokerCtx)
|
||||
.expect("ReliableChannelManager.new")
|
||||
|
||||
suite "Reliable Channel - ingress":
|
||||
asyncTest "manager dispatches marked WakuMessage to the right channel":
|
||||
## Unit test for the receive side of the API: instead of standing
|
||||
@ -54,7 +75,7 @@ suite "Reliable Channel - ingress":
|
||||
let appPayload = "hello reliable channel".toBytes()
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClient]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
@ -120,7 +141,7 @@ suite "Reliable Channel - ingress":
|
||||
let appPayload = "foreign payload".toBytes()
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClient]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
@ -165,8 +186,8 @@ suite "Reliable Channel - ingress":
|
||||
suite "Reliable Channel - send state machine":
|
||||
asyncTest "MessageSentEvent finalises the channelReqId as Sent":
|
||||
## Drives the real send pipeline (`send` -> segmentation -> SDS ->
|
||||
## rate_limit -> encrypt -> dispatch) via a fake `SendHandler` that
|
||||
## returns a canned `RequestId` instead of hitting the network.
|
||||
## rate_limit -> encrypt -> dispatch) via the `MessagingClientTest` double
|
||||
## that returns a canned `RequestId` instead of hitting the network.
|
||||
## Emitting the delivery-layer `MessageSentEvent` must drive the
|
||||
## channel-level state machine through `Confirmed` and produce a
|
||||
## `ChannelMessageSentEvent` (channel-level terminal event) for the
|
||||
@ -177,26 +198,24 @@ suite "Reliable Channel - send state machine":
|
||||
fakeMsgReqId = RequestId("fake-msg-req-1")
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClientTest]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
|
||||
manager = waku.reliableChannelManager
|
||||
manager = newTestManager(brokerCtx)
|
||||
|
||||
setNoopEncryption()
|
||||
|
||||
var sendCalls = 0
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
manager.messaging.send = proc(
|
||||
envelope: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} =
|
||||
sendCalls.inc
|
||||
return ok(fakeMsgReqId)
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
|
||||
)
|
||||
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
|
||||
.expect("createReliableChannel")
|
||||
|
||||
let sentFut = newFuture[RequestId]("channel-sent")
|
||||
@ -227,13 +246,14 @@ suite "Reliable Channel - send state machine":
|
||||
if finalised:
|
||||
check sentFut.read() == channelReqId
|
||||
|
||||
await manager.stop()
|
||||
(await waku.stop()).expect("stop")
|
||||
|
||||
asyncTest "two independent channelReqIds are finalised independently":
|
||||
## Two `send()` calls -> two independent `channelReqId`s, each with
|
||||
## one segment under the current segmentation skeleton
|
||||
## (`performSegmentation` always emits exactly one segment). The
|
||||
## fake `SendHandler` returns distinct `messagingReqId`s; finalising
|
||||
## send double returns distinct `messagingReqId`s; finalising
|
||||
## the first emits `ChannelMessageSentEvent` for its `channelReqId`,
|
||||
## finalising the second as a failure emits `ChannelMessageErrorEvent`
|
||||
## for the other.
|
||||
@ -242,27 +262,25 @@ suite "Reliable Channel - send state machine":
|
||||
contentTopic = ContentTopic("/reliable-channel/test/sm-multi")
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClientTest]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
|
||||
manager = waku.reliableChannelManager
|
||||
manager = newTestManager(brokerCtx)
|
||||
|
||||
setNoopEncryption()
|
||||
|
||||
var msgReqIds: seq[RequestId]
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
manager.messaging.send = proc(
|
||||
envelope: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} =
|
||||
let id = RequestId("fake-msg-req-" & $(msgReqIds.len + 1))
|
||||
msgReqIds.add(id)
|
||||
return ok(id)
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
|
||||
)
|
||||
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
|
||||
.expect("createReliableChannel")
|
||||
|
||||
let sentFut = newFuture[RequestId]("channel-sent")
|
||||
@ -319,6 +337,7 @@ suite "Reliable Channel - send state machine":
|
||||
if erroredArrived:
|
||||
check erroredFut.read() == channelReqId2
|
||||
|
||||
await manager.stop()
|
||||
(await waku.stop()).expect("stop")
|
||||
|
||||
asyncTest "TODO: channelReqId not pruned until ALL its segments are final":
|
||||
@ -328,11 +347,11 @@ suite "Reliable Channel - send state machine":
|
||||
## `channelReqId` cannot be produced through the real pipeline.
|
||||
## Implement once segmentation does real chunking: send a payload
|
||||
## larger than `DefaultSegmentSizeBytes`, capture the N
|
||||
## `messagingReqId`s from a fake `SendHandler`, finalise some, and
|
||||
## `messagingReqId`s from the send double, finalise some, and
|
||||
## assert prune only fires once every sibling is final.
|
||||
skip()
|
||||
|
||||
asyncTest "sibling MessageSentEvent during sendHandler await does not corrupt state":
|
||||
asyncTest "sibling MessageSentEvent during send await does not corrupt state":
|
||||
## Regression test for the prune-during-await race
|
||||
## (PR #3914 review comment r3324891059). Locks in that a sibling
|
||||
## `MessageSentEvent` firing while `onReadyToSend` is paused at an
|
||||
@ -343,20 +362,20 @@ suite "Reliable Channel - send state machine":
|
||||
contentTopic = ContentTopic("/reliable-channel/test/sm-race")
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClientTest]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
|
||||
manager = waku.reliableChannelManager
|
||||
manager = newTestManager(brokerCtx)
|
||||
|
||||
setNoopEncryption()
|
||||
|
||||
var msgReqIds: seq[RequestId]
|
||||
var sendsReturned = 0
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
manager.messaging.send = proc(
|
||||
envelope: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} =
|
||||
## 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
|
||||
@ -364,18 +383,15 @@ suite "Reliable Channel - send state machine":
|
||||
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: ""),
|
||||
MessageSentEvent.emit(
|
||||
brokerCtx, MessageSentEvent(requestId: msgReqIds[0], messageHash: "")
|
||||
)
|
||||
await sleepAsync(50.milliseconds)
|
||||
sendsReturned.inc()
|
||||
return ok(id)
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
|
||||
)
|
||||
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
|
||||
.expect("createReliableChannel")
|
||||
|
||||
var finalisedReqIds: seq[RequestId]
|
||||
@ -431,6 +447,7 @@ suite "Reliable Channel - send state machine":
|
||||
check channelReqId1 in finalisedReqIds
|
||||
check channelReqId2 in finalisedReqIds
|
||||
|
||||
await manager.stop()
|
||||
(await waku.stop()).expect("stop")
|
||||
|
||||
suite "Reliable Channel - SDS persistence":
|
||||
@ -450,22 +467,20 @@ suite "Reliable Channel - SDS persistence":
|
||||
removeDir(root)
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClientTest]
|
||||
lockNewGlobalBrokerContext:
|
||||
waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
|
||||
manager = waku.reliableChannelManager
|
||||
manager = newTestManager(globalBrokerContext())
|
||||
|
||||
setNoopEncryption()
|
||||
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
manager.messaging.send = proc(
|
||||
envelope: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} =
|
||||
return ok(RequestId("persist-msg-req-1"))
|
||||
|
||||
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")
|
||||
@ -496,6 +511,7 @@ suite "Reliable Channel - SDS persistence":
|
||||
check await pollMetaExists()
|
||||
check await pollLogRow()
|
||||
|
||||
await manager.stop()
|
||||
(await waku.stop()).expect("stop")
|
||||
|
||||
## A marked WakuMessage carrying an SDS envelope, as it arrives off the wire.
|
||||
@ -518,7 +534,7 @@ suite "Reliable Channel - SDS lifecycle":
|
||||
let payload2 = "second message".toBytes()
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClient]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
@ -589,7 +605,7 @@ suite "Reliable Channel - SDS lifecycle":
|
||||
let appPayload = "deliver once".toBytes()
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClient]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
@ -646,7 +662,7 @@ suite "Reliable Channel - SDS lifecycle":
|
||||
contentTopic = ContentTopic("/reliable-channel/test/foreign")
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClient]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
@ -706,7 +722,7 @@ suite "Reliable Channel - SDS lifecycle":
|
||||
removeDir(root)
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClient]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
@ -788,27 +804,25 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
contentTopic = ContentTopic("/reliable-channel/test/semantics")
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClientTest]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
|
||||
manager = waku.reliableChannelManager
|
||||
manager = newTestManager(brokerCtx)
|
||||
|
||||
setNoopEncryption()
|
||||
|
||||
var capturedWires: seq[seq[byte]]
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
manager.messaging.send = proc(
|
||||
envelope: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} =
|
||||
## Noop encryption is identity, so the envelope payload IS the SDS wire.
|
||||
capturedWires.add(env.payload)
|
||||
capturedWires.add(envelope.payload)
|
||||
return ok(RequestId("semantics-req-" & $capturedWires.len))
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
|
||||
)
|
||||
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
|
||||
.expect("createReliableChannel")
|
||||
|
||||
let remotePeer =
|
||||
@ -838,6 +852,7 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
check SdsMessageID("semantics-m1") in reply.causalHistory.getMessageIds()
|
||||
check reply.lamportTimestamp > m1.lamportTimestamp
|
||||
|
||||
await manager.stop()
|
||||
(await waku.stop()).expect("stop")
|
||||
|
||||
asyncTest "an unacknowledged send is acked by a later remote message":
|
||||
@ -856,26 +871,25 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
removeDir(root)
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClientTest]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
|
||||
manager = waku.reliableChannelManager
|
||||
manager = newTestManager(brokerCtx)
|
||||
|
||||
setNoopEncryption()
|
||||
|
||||
var capturedWires: seq[seq[byte]]
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
capturedWires.add(env.payload)
|
||||
manager.messaging.send = proc(
|
||||
envelope: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} =
|
||||
## Noop encryption is identity, so the envelope payload IS the SDS wire.
|
||||
capturedWires.add(envelope.payload)
|
||||
return ok(RequestId("ack-req-" & $capturedWires.len))
|
||||
|
||||
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")
|
||||
@ -932,6 +946,7 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
await sleepAsync(5.milliseconds)
|
||||
check bufLen == 0
|
||||
|
||||
await manager.stop()
|
||||
(await waku.stop()).expect("stop")
|
||||
|
||||
asyncTest "three-deep dependency chain is released in causal order":
|
||||
@ -944,7 +959,7 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
@["chain first".toBytes(), "chain second".toBytes(), "chain third".toBytes()]
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClient]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
@ -1019,7 +1034,7 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
let appPayload = "real message".toBytes()
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClient]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
@ -1092,26 +1107,25 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
let appPayload = "ok".toBytes()
|
||||
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClientTest]
|
||||
var brokerCtx: BrokerContext
|
||||
lockNewGlobalBrokerContext:
|
||||
brokerCtx = globalBrokerContext()
|
||||
waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
|
||||
manager = waku.reliableChannelManager
|
||||
manager = newTestManager(brokerCtx)
|
||||
|
||||
setNoopEncryption()
|
||||
|
||||
var capturedWires: seq[seq[byte]]
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
capturedWires.add(env.payload)
|
||||
manager.messaging.send = proc(
|
||||
envelope: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} =
|
||||
## Noop encryption is identity, so the envelope payload IS the SDS wire.
|
||||
capturedWires.add(envelope.payload)
|
||||
return ok(RequestId("unique-req-" & $capturedWires.len))
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
|
||||
)
|
||||
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
|
||||
.expect("createReliableChannel")
|
||||
|
||||
var deliveries: seq[seq[byte]]
|
||||
@ -1156,11 +1170,12 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
await sleepAsync(5.milliseconds)
|
||||
check deliveries.len == 2
|
||||
|
||||
await manager.stop()
|
||||
(await waku.stop()).expect("stop")
|
||||
|
||||
asyncTest "manager rejects operations on unknown channels":
|
||||
var waku: LogosDelivery
|
||||
var manager: ReliableChannelManager
|
||||
var manager: ReliableChannelManager[MessagingClient]
|
||||
lockNewGlobalBrokerContext:
|
||||
waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
|
||||
manager = waku.reliableChannelManager
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user