2021-01-06 09:35:05 +00:00
|
|
|
|
## Waku Relay module. Thin layer on top of GossipSub.
|
2020-04-29 05:02:42 +00:00
|
|
|
|
##
|
2021-01-06 09:35:05 +00:00
|
|
|
|
## See https://github.com/vacp2p/specs/blob/master/specs/waku/v2/waku-relay.md
|
|
|
|
|
## for spec.
|
2022-10-28 09:51:46 +00:00
|
|
|
|
when (NimMajor, NimMinor) < (1, 4):
|
|
|
|
|
{.push raises: [Defect].}
|
|
|
|
|
else:
|
|
|
|
|
{.push raises: [].}
|
2020-04-29 05:02:42 +00:00
|
|
|
|
|
2020-08-31 03:32:41 +00:00
|
|
|
|
import
|
2023-02-13 16:22:34 +00:00
|
|
|
|
std/[tables, sequtils],
|
2022-11-18 13:50:56 +00:00
|
|
|
|
stew/results,
|
2022-11-03 15:36:24 +00:00
|
|
|
|
chronos,
|
2023-01-11 08:10:46 +00:00
|
|
|
|
chronicles,
|
2022-11-03 15:36:24 +00:00
|
|
|
|
metrics,
|
2022-11-18 13:50:56 +00:00
|
|
|
|
libp2p/multihash,
|
2022-11-03 15:36:24 +00:00
|
|
|
|
libp2p/protocols/pubsub/pubsub,
|
2022-11-18 13:50:56 +00:00
|
|
|
|
libp2p/protocols/pubsub/rpc/messages,
|
2022-11-03 15:36:24 +00:00
|
|
|
|
libp2p/protocols/pubsub/gossipsub,
|
2023-01-11 08:10:46 +00:00
|
|
|
|
libp2p/stream/connection,
|
|
|
|
|
libp2p/switch
|
2022-11-09 08:55:47 +00:00
|
|
|
|
import
|
|
|
|
|
./waku_message
|
2020-08-31 03:32:41 +00:00
|
|
|
|
|
2020-05-01 05:00:00 +00:00
|
|
|
|
logScope:
|
2022-11-03 15:36:24 +00:00
|
|
|
|
topics = "waku relay"
|
2020-05-01 05:00:00 +00:00
|
|
|
|
|
2021-04-30 12:07:46 +00:00
|
|
|
|
const
|
2021-06-29 14:29:04 +00:00
|
|
|
|
WakuRelayCodec* = "/vac/waku/relay/2.0.0"
|
2020-05-01 05:00:00 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
|
|
|
|
|
type WakuRelayResult*[T] = Result[T, string]
|
|
|
|
|
|
|
|
|
|
type
|
|
|
|
|
PubsubRawHandler* = proc(pubsubTopic: PubsubTopic, data: seq[byte]): Future[void] {.gcsafe, raises: [Defect].}
|
2023-02-08 15:09:59 +00:00
|
|
|
|
SubscriptionHandler* = proc(pubsubTopic: PubsubTopic, message: WakuMessage): Future[void] {.gcsafe, raises: [Defect].}
|
2022-11-18 13:50:56 +00:00
|
|
|
|
|
2021-01-06 09:35:05 +00:00
|
|
|
|
type
|
|
|
|
|
WakuRelay* = ref object of GossipSub
|
|
|
|
|
|
2023-02-08 15:09:59 +00:00
|
|
|
|
WakuRelayHandler* = PubsubRawHandler|SubscriptionHandler
|
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
|
|
|
|
|
proc initProtocolHandler(w: WakuRelay) =
|
2020-05-11 04:05:28 +00:00
|
|
|
|
proc handler(conn: Connection, proto: string) {.async.} =
|
|
|
|
|
## main protocol handler that gets triggered on every
|
|
|
|
|
## connection for a protocol string
|
|
|
|
|
## e.g. ``/wakusub/0.0.1``, etc...
|
2022-11-18 13:50:56 +00:00
|
|
|
|
debug "Incoming WakuRelay connection", connection=conn, protocol=proto
|
2020-05-11 04:05:28 +00:00
|
|
|
|
|
2021-06-09 14:37:08 +00:00
|
|
|
|
try:
|
|
|
|
|
await w.handleConn(conn, proto)
|
|
|
|
|
except CancelledError:
|
|
|
|
|
# This is top-level procedure which will work as separate task, so it
|
|
|
|
|
# do not need to propogate CancelledError.
|
2022-11-18 13:50:56 +00:00
|
|
|
|
error "Unexpected cancellation in relay handler", conn=conn, error=getCurrentExceptionMsg()
|
|
|
|
|
except CatchableError:
|
|
|
|
|
error "WakuRelay handler leaks an error", conn=conn, error=getCurrentExceptionMsg()
|
2020-05-11 04:05:28 +00:00
|
|
|
|
|
2020-06-01 03:42:16 +00:00
|
|
|
|
# XXX: Handler hijack GossipSub here?
|
2020-05-11 04:05:28 +00:00
|
|
|
|
w.handler = handler
|
2020-08-26 11:28:24 +00:00
|
|
|
|
w.codec = WakuRelayCodec
|
2020-05-11 04:05:28 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
method initPubSub(w: WakuRelay) {.raises: [InitializationError].} =
|
2023-01-11 08:10:46 +00:00
|
|
|
|
## NOTE: This method overrides GossipSub initPubSub method; it called by the
|
2022-11-18 13:50:56 +00:00
|
|
|
|
## parent protocol, PubSub.
|
|
|
|
|
debug "init waku relay"
|
2020-06-02 11:27:53 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
# After discussions with @sinkingsugar: This is essentially what is needed for
|
2020-12-17 16:06:59 +00:00
|
|
|
|
# the libp2p `StrictNoSign` policy
|
|
|
|
|
w.anonymize = true
|
|
|
|
|
w.verifySignature = false
|
|
|
|
|
w.sign = false
|
|
|
|
|
|
2020-11-24 09:14:18 +00:00
|
|
|
|
procCall GossipSub(w).initPubSub()
|
2020-06-02 11:27:53 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
w.initProtocolHandler()
|
|
|
|
|
|
|
|
|
|
|
2023-02-10 14:17:50 +00:00
|
|
|
|
proc new*(T: type WakuRelay, switch: Switch, triggerSelf: bool = true): WakuRelayResult[T] =
|
2023-01-11 08:10:46 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
proc msgIdProvider(msg: messages.Message): Result[MessageID, ValidationResult] =
|
2023-02-13 16:22:34 +00:00
|
|
|
|
let hash = sha256.digest(msg.data)
|
|
|
|
|
ok(toSeq(hash.data))
|
2022-11-18 13:50:56 +00:00
|
|
|
|
|
|
|
|
|
var wr: WakuRelay
|
|
|
|
|
try:
|
|
|
|
|
wr = WakuRelay.init(
|
2023-01-11 08:10:46 +00:00
|
|
|
|
switch = switch,
|
2022-11-18 13:50:56 +00:00
|
|
|
|
msgIdProvider = msgIdProvider,
|
|
|
|
|
triggerSelf = triggerSelf,
|
|
|
|
|
sign = false,
|
|
|
|
|
verifySignature = false,
|
|
|
|
|
maxMessageSize = MaxWakuMessageSize
|
|
|
|
|
)
|
|
|
|
|
except InitializationError:
|
|
|
|
|
return err("initialization error: " & getCurrentExceptionMsg())
|
|
|
|
|
|
2023-02-10 14:17:50 +00:00
|
|
|
|
# TODO: Add a function to validate the WakuMessage integrity
|
|
|
|
|
# # Rejects messages that are not WakuMessage
|
|
|
|
|
# proc validator(topic: string, message: messages.Message): Future[ValidationResult] {.async.} =
|
|
|
|
|
# let msg = WakuMessage.decode(message.data)
|
|
|
|
|
# if msg.isOk():
|
|
|
|
|
# return ValidationResult.Accept
|
|
|
|
|
# return ValidationResult.Reject
|
|
|
|
|
|
|
|
|
|
# # Add validator to all default pubsub topics
|
|
|
|
|
# for pubSubTopic in defaultPubsubTopics:
|
|
|
|
|
# wr.addValidator(pubSubTopic, validator)
|
2022-11-18 13:50:56 +00:00
|
|
|
|
|
|
|
|
|
ok(wr)
|
|
|
|
|
|
|
|
|
|
|
2023-02-10 14:17:50 +00:00
|
|
|
|
method addValidator*(w: WakuRelay, topic: varargs[string], handler: ValidatorHandler) {.gcsafe.} =
|
|
|
|
|
procCall GossipSub(w).addValidator(topic, handler)
|
|
|
|
|
|
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
method start*(w: WakuRelay) {.async.} =
|
|
|
|
|
debug "start"
|
|
|
|
|
await procCall GossipSub(w).start()
|
|
|
|
|
|
|
|
|
|
method stop*(w: WakuRelay) {.async.} =
|
|
|
|
|
debug "stop"
|
|
|
|
|
await procCall GossipSub(w).stop()
|
2020-05-11 04:05:28 +00:00
|
|
|
|
|
2020-11-24 09:14:18 +00:00
|
|
|
|
|
2023-02-08 15:09:59 +00:00
|
|
|
|
proc isSubscribed*(w: WakuRelay, topic: PubsubTopic): bool =
|
|
|
|
|
GossipSub(w).topics.hasKey(topic)
|
|
|
|
|
|
|
|
|
|
iterator subscribedTopics*(w: WakuRelay): lent PubsubTopic =
|
|
|
|
|
for topic in GossipSub(w).topics.keys():
|
|
|
|
|
yield topic
|
|
|
|
|
|
|
|
|
|
method subscribe*(w: WakuRelay, pubsubTopic: PubsubTopic, handler: WakuRelayHandler) =
|
2022-11-18 13:50:56 +00:00
|
|
|
|
debug "subscribe", pubsubTopic=pubsubTopic
|
2020-05-22 07:35:31 +00:00
|
|
|
|
|
2023-01-11 08:10:46 +00:00
|
|
|
|
var subsHandler: PubsubRawHandler
|
2023-02-08 15:09:59 +00:00
|
|
|
|
when handler is SubscriptionHandler:
|
|
|
|
|
subsHandler = proc(pubsubTopic: PubsubTopic, data: seq[byte]): Future[void] {.gcsafe.} =
|
2022-11-18 13:50:56 +00:00
|
|
|
|
let decodeRes = WakuMessage.decode(data)
|
|
|
|
|
if decodeRes.isErr():
|
|
|
|
|
debug "message decode failure", pubsubTopic=pubsubTopic, error=decodeRes.error
|
|
|
|
|
return
|
2020-06-02 11:27:53 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
handler(pubsubTopic, decodeRes.value)
|
|
|
|
|
else:
|
|
|
|
|
subsHandler = handler
|
2020-05-15 03:29:01 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
procCall GossipSub(w).subscribe(pubsubTopic, subsHandler)
|
|
|
|
|
|
|
|
|
|
method unsubscribe*(w: WakuRelay, topics: seq[TopicPair]) =
|
2023-02-08 15:09:59 +00:00
|
|
|
|
debug "unsubscribe", pubsubTopic=topics.mapIt(it[0])
|
2020-11-24 09:14:18 +00:00
|
|
|
|
|
2021-02-02 11:33:59 +00:00
|
|
|
|
procCall GossipSub(w).unsubscribe(topics)
|
2020-06-01 03:42:16 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
method unsubscribeAll*(w: WakuRelay, pubsubTopic: PubsubTopic) =
|
|
|
|
|
debug "unsubscribeAll", pubsubTopic=pubsubTopic
|
2020-11-24 09:14:18 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
procCall GossipSub(w).unsubscribeAll(pubsubTopic)
|
2020-10-27 01:13:56 +00:00
|
|
|
|
|
2020-06-01 03:42:16 +00:00
|
|
|
|
|
2022-11-18 13:50:56 +00:00
|
|
|
|
method publish*(w: WakuRelay, pubsubTopic: PubsubTopic, message: WakuMessage|seq[byte]): Future[int] {.async.} =
|
|
|
|
|
trace "publish", pubsubTopic=pubsubTopic
|
|
|
|
|
|
|
|
|
|
var data: seq[byte]
|
|
|
|
|
when message is WakuMessage:
|
2023-01-25 13:35:38 +00:00
|
|
|
|
data = message.encode().buffer
|
2022-11-18 13:50:56 +00:00
|
|
|
|
else:
|
|
|
|
|
data = message
|
|
|
|
|
|
2023-01-25 13:35:38 +00:00
|
|
|
|
return await procCall GossipSub(w).publish(pubsubTopic, data)
|
2022-11-18 13:50:56 +00:00
|
|
|
|
|
2021-04-30 12:07:46 +00:00
|
|
|
|
|