diff --git a/apps/chat2/chat2.nim b/apps/chat2/chat2.nim index 2e57fc015..817b8ae15 100644 --- a/apps/chat2/chat2.nim +++ b/apps/chat2/chat2.nim @@ -27,11 +27,12 @@ import peerinfo, # manage the information of a peer, such as peer ID and public / private key peerid, # Implement how peers interact - protobuf/minprotobuf, # message serialisation/deserialisation from and to protobufs nameresolving/dnsresolver, ] # define DNS resolution +import protobuf_serialization, protobuf_serialization/pkg/results import logos_delivery/waku/[ + common/protobuf, waku_core, waku_lightpush_legacy/common, waku_lightpush_legacy/rpc, @@ -87,27 +88,36 @@ type nick*: string payload*: seq[byte] -proc init*(T: type Chat2Message, buffer: seq[byte]): ProtoResult[T] = - var msg = Chat2Message() - let pb = initProtoBuffer(buffer) +type Chat2MessagePB {.proto2.} = object + timestamp {.fieldNumber: 1, pint.}: Opt[uint64] + nick {.fieldNumber: 2.}: Opt[string] + payload {.fieldNumber: 3.}: Opt[seq[byte]] - var timestamp: uint64 - discard ?pb.getField(1, timestamp) - msg.timestamp = int64(timestamp) +proc decodeChat2Message(buffer: seq[byte]): ProtobufResult[Chat2Message] = + var pb: Chat2MessagePB + try: + pb = Protobuf.decode(buffer, Chat2MessagePB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + ok( + Chat2Message( + timestamp: int64(pb.timestamp.get(0'u64)), + nick: pb.nick.get(""), + payload: pb.payload.get(@[]), + ) + ) - discard ?pb.getField(2, msg.nick) - discard ?pb.getField(3, msg.payload) +proc init*(T: type Chat2Message, buffer: seq[byte]): ProtobufResult[T] = + decodeChat2Message(buffer) - ok(msg) - -proc encode*(message: Chat2Message): ProtoBuffer = - var serialised = initProtoBuffer() - - serialised.write(1, uint64(message.timestamp)) - serialised.write(2, message.nick) - serialised.write(3, message.payload) - - return serialised +proc encode*(message: Chat2Message): seq[byte] = + Protobuf.encode( + Chat2MessagePB( + timestamp: Opt.some(uint64(message.timestamp)), + nick: Opt.some(message.nick), + payload: Opt.some(message.payload), + ) + ) proc toString*(message: Chat2Message): string = # Get message date and timestamp in local time @@ -183,7 +193,7 @@ proc publish(c: Chat, line: string) = trace "lightpush response received", response = response var message = WakuMessage( - payload: chat2pb.buffer, + payload: chat2pb, contentTopic: c.contentTopic, version: 0, timestamp: getNanosecondTime(time), diff --git a/apps/chat2bridge/chat2bridge.nim b/apps/chat2bridge/chat2bridge.nim index 097735c7a..a6ddad623 100644 --- a/apps/chat2bridge/chat2bridge.nim +++ b/apps/chat2bridge/chat2bridge.nim @@ -89,7 +89,7 @@ proc toWakuMessage( payload: msgFields["text"].getStr().toBytes(), ).encode() - WakuMessage(payload: chat2pb.buffer, contentTopic: cmb.contentTopic, version: 0) + WakuMessage(payload: chat2pb, contentTopic: cmb.contentTopic, version: 0) proc toChat2(cmb: Chat2MatterBridge, jsonNode: JsonNode) {.async.} = let msg = cmb.toWakuMessage(jsonNode) diff --git a/apps/chat2mix/chat2mix.nim b/apps/chat2mix/chat2mix.nim index 4d2d08837..e67243a2a 100644 --- a/apps/chat2mix/chat2mix.nim +++ b/apps/chat2mix/chat2mix.nim @@ -27,14 +27,15 @@ import peerinfo, # manage the information of a peer, such as peer ID and public / private key peerid, # Implement how peers interact - protobuf/minprotobuf, # message serialisation/deserialisation from and to protobufs protocols/kademlia/types, protocols/service_discovery/types as sd_types, nameresolving/dnsresolver, ] # define DNS resolution import libp2p_mix/[curve25519, mix_protocol] +import protobuf_serialization, protobuf_serialization/pkg/results import logos_delivery/waku/[ + common/protobuf, waku_core, waku_core/peers, waku_lightpush/common, @@ -108,27 +109,36 @@ proc getPubsubTopic*( return "" #TODO: fix this. return $RelayShard(clusterId: conf.clusterId, shardId: shard.shardId) -proc init*(T: type Chat2Message, buffer: seq[byte]): ProtoResult[T] = - var msg = Chat2Message() - let pb = initProtoBuffer(buffer) +type Chat2MessagePB {.proto2.} = object + timestamp {.fieldNumber: 1, pint.}: Opt[uint64] + nick {.fieldNumber: 2.}: Opt[string] + payload {.fieldNumber: 3.}: Opt[seq[byte]] - var timestamp: uint64 - discard ?pb.getField(1, timestamp) - msg.timestamp = int64(timestamp) +proc decodeChat2Message(buffer: seq[byte]): ProtobufResult[Chat2Message] = + var pb: Chat2MessagePB + try: + pb = Protobuf.decode(buffer, Chat2MessagePB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + ok( + Chat2Message( + timestamp: int64(pb.timestamp.get(0'u64)), + nick: pb.nick.get(""), + payload: pb.payload.get(@[]), + ) + ) - discard ?pb.getField(2, msg.nick) - discard ?pb.getField(3, msg.payload) +proc init*(T: type Chat2Message, buffer: seq[byte]): ProtobufResult[T] = + decodeChat2Message(buffer) - ok(msg) - -proc encode*(message: Chat2Message): ProtoBuffer = - var serialised = initProtoBuffer() - - serialised.write(1, uint64(message.timestamp)) - serialised.write(2, message.nick) - serialised.write(3, message.payload) - - return serialised +proc encode*(message: Chat2Message): seq[byte] = + Protobuf.encode( + Chat2MessagePB( + timestamp: Opt.some(uint64(message.timestamp)), + nick: Opt.some(message.nick), + payload: Opt.some(message.payload), + ) + ) proc `$`*(message: Chat2Message): string = # Get message date and timestamp in local time @@ -203,7 +213,7 @@ proc publish(c: Chat, line: string) {.async.} = trace "lightpush response received", response = response var message = WakuMessage( - payload: chat2pb.buffer, + payload: chat2pb, contentTopic: c.contentTopic, version: 0, timestamp: getNanosecondTime(time), diff --git a/examples/wakustealthcommitments/wire_spec.nim b/examples/wakustealthcommitments/wire_spec.nim index fdaa4d8f3..9cd7ff7d0 100644 --- a/examples/wakustealthcommitments/wire_spec.nim +++ b/examples/wakustealthcommitments/wire_spec.nim @@ -1,10 +1,10 @@ import std/times import confutils, chronicles, chronos, results +import protobuf_serialization, protobuf_serialization/pkg/results import logos_delivery/waku/[waku_core, common/protobuf] -import libp2p/protobuf/minprotobuf -export times, confutils, chronicles, chronos, results, waku_core, protobuf, minprotobuf +export times, confutils, chronicles, chronos, results, waku_core, protobuf type SerializedKey* = seq[byte] @@ -16,57 +16,47 @@ type WakuStealthCommitmentMsg* = object stealthCommitment*: Opt[SerializedKey] viewTag*: Opt[uint64] -proc decode*(T: type WakuStealthCommitmentMsg, buffer: seq[byte]): ProtoResult[T] = - var msg = WakuStealthCommitmentMsg() - let pb = initProtoBuffer(buffer) +type WakuStealthCommitmentMsgPB {.proto2.} = object + request {.fieldNumber: 1, pint.}: Opt[uint64] + spendingPubKey {.fieldNumber: 2.}: Opt[seq[byte]] + viewingPubKey {.fieldNumber: 3.}: Opt[seq[byte]] + stealthCommitment {.fieldNumber: 4.}: Opt[seq[byte]] + ephemeralPubKey {.fieldNumber: 5.}: Opt[seq[byte]] + viewTag {.fieldNumber: 6, pint.}: Opt[uint64] - var request: uint64 - discard ?pb.getField(1, request) - msg.request = request == 1 - var spendingPubKey = newSeq[byte]() - discard ?pb.getField(2, spendingPubKey) - msg.spendingPubKey = - if spendingPubKey.len > 0: - Opt.some(spendingPubKey) - else: - Opt.none(SerializedKey) - var viewingPubKey = newSeq[byte]() - discard ?pb.getField(3, viewingPubKey) - msg.viewingPubKey = - if viewingPubKey.len > 0: - Opt.some(viewingPubKey) - else: - Opt.none(SerializedKey) +proc nonEmptyKey(o: Opt[seq[byte]]): Opt[SerializedKey] = + if o.isSome() and o.get().len > 0: + o + else: + Opt.none(SerializedKey) + +proc decode*(T: type WakuStealthCommitmentMsg, buffer: seq[byte]): ProtobufResult[T] = + var pb: WakuStealthCommitmentMsgPB + try: + pb = Protobuf.decode(buffer, WakuStealthCommitmentMsgPB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + + var msg = WakuStealthCommitmentMsg() + msg.request = pb.request.get(0'u64) == 1 + msg.spendingPubKey = nonEmptyKey(pb.spendingPubKey) + msg.viewingPubKey = nonEmptyKey(pb.viewingPubKey) if msg.spendingPubKey.isSome() and msg.viewingPubKey.isSome(): msg.stealthCommitment = Opt.none(SerializedKey) msg.viewTag = Opt.none(uint64) return ok(msg) if msg.spendingPubKey.isSome() and msg.viewingPubKey.isNone(): - return err(ProtoError.RequiredFieldMissing) + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) if msg.spendingPubKey.isNone() and msg.viewingPubKey.isSome(): - return err(ProtoError.RequiredFieldMissing) + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) if msg.request == true and msg.spendingPubKey.isNone() and msg.viewingPubKey.isNone(): - return err(ProtoError.RequiredFieldMissing) + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) - var stealthCommitment = newSeq[byte]() - discard ?pb.getField(4, stealthCommitment) - msg.stealthCommitment = - if stealthCommitment.len > 0: - Opt.some(stealthCommitment) - else: - Opt.none(SerializedKey) + msg.stealthCommitment = nonEmptyKey(pb.stealthCommitment) + msg.ephemeralPubKey = nonEmptyKey(pb.ephemeralPubKey) - var ephemeralPubKey = newSeq[byte]() - discard ?pb.getField(5, ephemeralPubKey) - msg.ephemeralPubKey = - if ephemeralPubKey.len > 0: - Opt.some(ephemeralPubKey) - else: - Opt.none(SerializedKey) - - var viewTag: uint64 - discard ?pb.getField(6, viewTag) + let viewTag = pb.viewTag.get(0'u64) msg.viewTag = if viewTag != 0: Opt.some(viewTag) @@ -75,13 +65,13 @@ proc decode*(T: type WakuStealthCommitmentMsg, buffer: seq[byte]): ProtoResult[T if msg.stealthCommitment.isNone() and msg.viewTag.isNone() and msg.ephemeralPubKey.isNone(): - return err(ProtoError.RequiredFieldMissing) + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) if msg.stealthCommitment.isSome() and msg.viewTag.isNone(): - return err(ProtoError.RequiredFieldMissing) + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) if msg.stealthCommitment.isNone() and msg.viewTag.isSome(): - return err(ProtoError.RequiredFieldMissing) + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) if msg.stealthCommitment.isSome() and msg.viewTag.isSome(): msg.spendingPubKey = Opt.none(SerializedKey) @@ -89,23 +79,17 @@ proc decode*(T: type WakuStealthCommitmentMsg, buffer: seq[byte]): ProtoResult[T ok(msg) -proc encode*(msg: WakuStealthCommitmentMsg): ProtoBuffer = - var serialised = initProtoBuffer() - - serialised.write(1, uint64(msg.request)) - - if msg.spendingPubKey.isSome(): - serialised.write(2, msg.spendingPubKey.get()) - if msg.viewingPubKey.isSome(): - serialised.write(3, msg.viewingPubKey.get()) - if msg.stealthCommitment.isSome(): - serialised.write(4, msg.stealthCommitment.get()) - if msg.ephemeralPubKey.isSome(): - serialised.write(5, msg.ephemeralPubKey.get()) - if msg.viewTag.isSome(): - serialised.write(6, msg.viewTag.get()) - - return serialised +proc encode*(msg: WakuStealthCommitmentMsg): seq[byte] = + Protobuf.encode( + WakuStealthCommitmentMsgPB( + request: Opt.some(uint64(msg.request)), + spendingPubKey: msg.spendingPubKey, + viewingPubKey: msg.viewingPubKey, + stealthCommitment: msg.stealthCommitment, + ephemeralPubKey: msg.ephemeralPubKey, + viewTag: msg.viewTag, + ) + ) func toByteSeq*(str: string): seq[byte] {.inline.} = ## Converts a string to the corresponding byte sequence. diff --git a/logos_delivery.nimble b/logos_delivery.nimble index 8c4d32cb5..fa8130edf 100644 --- a/logos_delivery.nimble +++ b/logos_delivery.nimble @@ -26,6 +26,7 @@ requires "nim >= 2.2.4", "serialization", "json_serialization", "toml_serialization", + "protobuf_serialization", "faststreams", # Networking & P2P "https://github.com/vacp2p/nim-libp2p.git#v2.0.0", diff --git a/logos_delivery/waku/common/protobuf.nim b/logos_delivery/waku/common/protobuf.nim index a47fb7435..d8b48fadf 100644 --- a/logos_delivery/waku/common/protobuf.nim +++ b/logos_delivery/waku/common/protobuf.nim @@ -1,10 +1,9 @@ -# Extensions for libp2p's protobuf library implementation +# Waku protobuf result/error types. {.push raises: [].} -import results, libp2p/protobuf/minprotobuf, libp2p/varint - -export minprotobuf, varint +import results +export results ## Custom errors @@ -17,48 +16,23 @@ type ProtobufError* = object case kind*: ProtobufErrorKind of DecodeFailure: - error*: minprotobuf.ProtoError + discard of MissingRequiredField, InvalidLengthField: field*: string ProtobufResult*[T] = Result[T, ProtobufError] -converter toProtobufError*(err: minprotobuf.ProtoError): ProtobufError = - case err - of minprotobuf.ProtoError.RequiredFieldMissing: - ProtobufError(kind: ProtobufErrorKind.MissingRequiredField, field: "unknown") - else: - ProtobufError(kind: ProtobufErrorKind.DecodeFailure, error: err) - proc missingRequiredField*(T: type ProtobufError, field: string): T = ProtobufError(kind: ProtobufErrorKind.MissingRequiredField, field: field) proc invalidLengthField*(T: type ProtobufError, field: string): T = ProtobufError(kind: ProtobufErrorKind.InvalidLengthField, field: field) -## Extension methods - -proc write3*(proto: var ProtoBuffer, field: int, value: auto) = - when value is Opt: - if value.isSome(): - proto.write(field, value.get()) - else: - proto.write(field, value) - -proc finish3*(proto: var ProtoBuffer) = - if proto.buffer.len > 0: - proto.finish() - else: - proto.offset = 0 - -proc `==`*(a: zint64, b: zint64): bool = - int64(a) == int64(b) - proc `$`*(err: ProtobufError): string = case err.kind of DecodeFailure: - return $err.error ## assume that ProtoError is pure + "DecodeFailure" of MissingRequiredField: - return "MissingRequiredField " & err.field + "MissingRequiredField " & err.field of InvalidLengthField: - return "InvalidLengthField " & err.field + "InvalidLengthField " & err.field diff --git a/logos_delivery/waku/common/protobuf_ext.nim b/logos_delivery/waku/common/protobuf_ext.nim new file mode 100644 index 000000000..ee0b044a1 --- /dev/null +++ b/logos_delivery/waku/common/protobuf_ext.nim @@ -0,0 +1,49 @@ +## array[N, byte] serialized as a length-delimited bytes field. + +{.push raises: [].} + +import faststreams +import protobuf_serialization +import protobuf_serialization/pkg/results +import protobuf_serialization/std/enums + +export protobuf_serialization, results, enums + +func supportsPacked*[N: static int]( + T: type array[N, byte], ProtoType: type ProtobufExt +): bool = + false + +func computeFieldSize*[N: static int]( + field: int, + value: array[N, byte], + ProtoType: type ProtobufExt, + skipDefault: static bool, +): int = + computeFieldSize(field, @value, pbytes, skipDefault) + +proc writeField*[N: static int]( + stream: OutputStream, + field: int, + value: array[N, byte], + ProtoType: type ProtobufExt, + skipDefault: static bool = false, +) {.raises: [IOError].} = + writeField(stream, field, @value, pbytes, skipDefault) + +proc readFieldInto*[N: static int]( + stream: InputStream, + value: var array[N, byte], + header: FieldHeader, + ProtoType: type ProtobufExt, +): bool {.raises: [SerializationError, IOError].} = + var s: seq[byte] + if readFieldInto(stream, s, header, pbytes): + if s.len == N: + for i in 0 ..< N: + value[i] = s[i] + true + else: + false + else: + false diff --git a/logos_delivery/waku/incentivization/rpc.nim b/logos_delivery/waku/incentivization/rpc.nim index 6e54ec201..d5e1b83d3 100644 --- a/logos_delivery/waku/incentivization/rpc.nim +++ b/logos_delivery/waku/incentivization/rpc.nim @@ -1,12 +1,12 @@ -import results +import results, protobuf_serialization, protobuf_serialization/pkg/results # Implementing the RFC: # https://github.com/vacp2p/rfc/tree/master/content/docs/rfcs/73 type - EligibilityProof* = object - proofOfPayment*: Opt[seq[byte]] + EligibilityProof* {.proto2.} = object + proofOfPayment* {.fieldNumber: 1.}: Opt[seq[byte]] - EligibilityStatus* = object - statusCode*: uint32 - statusDesc*: Opt[string] + EligibilityStatus* {.proto2.} = object + statusCode* {.fieldNumber: 1, pint, required.}: uint32 + statusDesc* {.fieldNumber: 2.}: Opt[string] diff --git a/logos_delivery/waku/incentivization/rpc_codec.nim b/logos_delivery/waku/incentivization/rpc_codec.nim index 1696bb8c3..dbc997dd0 100644 --- a/logos_delivery/waku/incentivization/rpc_codec.nim +++ b/logos_delivery/waku/incentivization/rpc_codec.nim @@ -1,50 +1,30 @@ -import results, ../common/protobuf, ./rpc +import protobuf_serialization, protobuf_serialization/pkg/results +import ../common/protobuf, ./rpc # Codec for EligibilityProof -proc encode*(epRpc: EligibilityProof): ProtoBuffer = - var pb = initProtoBuffer() - if epRpc.proofOfPayment.isSome(): - let proofOfPayment = epRpc.proofOfPayment.get() - pb.write3(1, proofOfPayment) - else: - # there is no proof - discard - pb +proc encode*(epRpc: EligibilityProof): seq[byte] = + Protobuf.encode(epRpc) + +proc decodeEligibilityProof(buffer: seq[byte]): ProtobufResult[EligibilityProof] = + try: + ok(Protobuf.decode(buffer, EligibilityProof)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) proc decode*(T: type EligibilityProof, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var epRpc = EligibilityProof() - var proofOfPayment = newSeq[byte]() - if not ?pb.getField(1, proofOfPayment): - epRpc.proofOfPayment = Opt.none(seq[byte]) - else: - epRpc.proofOfPayment = Opt.some(proofOfPayment) - ok(epRpc) + decodeEligibilityProof(buffer) # Codec for EligibilityStatus -proc encode*(esRpc: EligibilityStatus): ProtoBuffer = - var pb = initProtoBuffer() - pb.write3(1, esRpc.statusCode) - if esRpc.statusDesc.isSome(): - pb.write3(2, esRpc.statusDesc.get()) - pb +proc encode*(esRpc: EligibilityStatus): seq[byte] = + Protobuf.encode(esRpc) + +proc decodeEligibilityStatus(buffer: seq[byte]): ProtobufResult[EligibilityStatus] = + try: + ok(Protobuf.decode(buffer, EligibilityStatus)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) proc decode*(T: type EligibilityStatus, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var esRpc = EligibilityStatus() - # status code - var code = uint32(0) - if not ?pb.getField(1, code): - # status code is mandatory - return err(ProtobufError.missingRequiredField("status_code")) - else: - esRpc.statusCode = code - # status description - var description = "" - if not ?pb.getField(2, description): - esRpc.statusDesc = Opt.none(string) - else: - esRpc.statusDesc = Opt.some(description) - ok(esRpc) + decodeEligibilityStatus(buffer) diff --git a/logos_delivery/waku/node/peer_manager/peer_store/waku_peer_storage.nim b/logos_delivery/waku/node/peer_manager/peer_store/waku_peer_storage.nim index f52fd6466..63b0f934a 100644 --- a/logos_delivery/waku/node/peer_manager/peer_store/waku_peer_storage.nim +++ b/logos_delivery/waku/node/peer_manager/peer_store/waku_peer_storage.nim @@ -1,9 +1,11 @@ {.push raises: [].} -import - std/sets, results, sqlite3_abi, eth/p2p/discoveryv5/enr, libp2p/protobuf/minprotobuf +import std/sets, results, sqlite3_abi, eth/p2p/discoveryv5/enr +import protobuf_serialization, protobuf_serialization/pkg/results +import libp2p/[peerid, multiaddress, crypto/crypto] import ../../../common/databases/db_sqlite, + ../../../common/protobuf, ../../../waku_core, ../waku_peer_store, ./peer_storage @@ -18,62 +20,67 @@ type WakuPeerStorage* = ref object of PeerStorage # Protobuf Serialisation # ########################## -proc decode*(T: type RemotePeerInfo, buffer: seq[byte]): ProtoResult[T] = - var - multiaddrSeq: seq[MultiAddress] - protoSeq: seq[string] - storedInfo = RemotePeerInfo() - rlpBytes: seq[byte] - connectedness: uint32 - disconnectTime: uint64 +type RemotePeerInfoPB {.proto2.} = object + peerId {.fieldNumber: 1, ext, required.}: PeerId + addrs {.fieldNumber: 2, ext.}: seq[MultiAddress] + protocols {.fieldNumber: 3.}: seq[string] + publicKey {.fieldNumber: 4, required.}: seq[byte] + connectedness {.fieldNumber: 5, pint.}: Opt[uint32] + disconnectTime {.fieldNumber: 6, pint.}: Opt[uint64] + enr {.fieldNumber: 7.}: Opt[seq[byte]] - var pb = initProtoBuffer(buffer) +proc decodeRemotePeerInfo(buffer: seq[byte]): ProtobufResult[RemotePeerInfo] = + var pb: RemotePeerInfoPB + try: + pb = Protobuf.decode(buffer, RemotePeerInfoPB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) - discard ?pb.getField(1, storedInfo.peerId) - discard ?pb.getRepeatedField(2, multiaddrSeq) - discard ?pb.getRepeatedField(3, protoSeq) - discard ?pb.getField(4, storedInfo.publicKey) - discard ?pb.getField(5, connectedness) - discard ?pb.getField(6, disconnectTime) - let hasENR = ?pb.getField(7, rlpBytes) + var storedInfo = RemotePeerInfo() + storedInfo.peerId = pb.peerId + storedInfo.addrs = pb.addrs + storedInfo.protocols = pb.protocols - storedInfo.addrs = multiaddrSeq - storedInfo.protocols = protoSeq - storedInfo.connectedness = Connectedness(connectedness) - storedInfo.disconnectTime = int64(disconnectTime) + var publicKey: crypto.PublicKey + if publicKey.init(pb.publicKey): + storedInfo.publicKey = publicKey - if hasENR: + storedInfo.connectedness = Connectedness(pb.connectedness.get(0'u32)) + storedInfo.disconnectTime = int64(pb.disconnectTime.get(0'u64)) + + if pb.enr.isSome(): var record: Record - - if record.fromBytes(rlpBytes): + if record.fromBytes(pb.enr.get()): storedInfo.enr = Opt.some(record) ok(storedInfo) -proc encode*(remotePeerInfo: RemotePeerInfo): PeerStorageResult[ProtoBuffer] = - var pb = initProtoBuffer() +proc decode*(T: type RemotePeerInfo, buffer: seq[byte]): ProtobufResult[T] = + decodeRemotePeerInfo(buffer) - pb.write(1, remotePeerInfo.peerId) +proc encode*(remotePeerInfo: RemotePeerInfo): PeerStorageResult[seq[byte]] = + let publicKeyBytes = remotePeerInfo.publicKey.getBytes().valueOr: + return err("Encoding public key failed: " & $error) - for multiaddr in remotePeerInfo.addrs.items: - pb.write(2, multiaddr) + let enr = + if remotePeerInfo.enr.isSome(): + Opt.some(remotePeerInfo.enr.get().raw) + else: + Opt.none(seq[byte]) - for proto in remotePeerInfo.protocols.items: - pb.write(3, proto) - - let catchRes = catch: - pb.write(4, remotePeerInfo.publicKey) - catchRes.isOkOr: - return err("Enncoding public key failed: " & catchRes.error.msg) - - pb.write(5, uint32(ord(remotePeerInfo.connectedness))) - - pb.write(6, uint64(remotePeerInfo.disconnectTime)) - - if remotePeerInfo.enr.isSome(): - pb.write(7, remotePeerInfo.enr.get().raw) - - return ok(pb) + ok( + Protobuf.encode( + RemotePeerInfoPB( + peerId: remotePeerInfo.peerId, + addrs: remotePeerInfo.addrs, + protocols: remotePeerInfo.protocols, + publicKey: publicKeyBytes, + connectedness: Opt.some(uint32(ord(remotePeerInfo.connectedness))), + disconnectTime: Opt.some(uint64(remotePeerInfo.disconnectTime)), + enr: enr, + ) + ) + ) ########################## # Storage implementation # @@ -128,7 +135,7 @@ method put*( let encoded = remotePeerInfo.encode().valueOr: return err("peer info encoding failed: " & error) - db.replaceStmt.exec((remotePeerInfo.peerId.data, encoded.buffer)).isOkOr: + db.replaceStmt.exec((remotePeerInfo.peerId.data, encoded)).isOkOr: return err("DB operation failed: " & error) return ok() @@ -138,7 +145,9 @@ method getAll*( ): PeerStorageResult[void] = ## Retrieves all peers from storage - proc peer(s: ptr sqlite3_stmt) {.gcsafe, raises: [ResultError[ProtoError]].} = + proc peer( + s: ptr sqlite3_stmt + ) {.gcsafe, raises: [ResultError[protobuf.ProtobufError]].} = let # Stored Info sTo = cast[ptr UncheckedArray[byte]](sqlite3_column_blob(s, 1)) diff --git a/logos_delivery/waku/rln/proof.nim b/logos_delivery/waku/rln/proof.nim index ba4b0361a..5e7e4689d 100644 --- a/logos_delivery/waku/rln/proof.nim +++ b/logos_delivery/waku/rln/proof.nim @@ -69,7 +69,7 @@ proc generateRLNProof*( try: let proof = (await rln.groupManager.generateProof(input, epoch, nonce)).valueOr: return err("could not generate rln-v2 proof: " & $error) - return ok(proof.encode().buffer) + return ok(proof.encode()) except CatchableError as e: return err("exception generating rln proof: " & e.msg) diff --git a/logos_delivery/waku/rln/protocol_types.nim b/logos_delivery/waku/rln/protocol_types.nim index 4a34d57ed..1348b9881 100644 --- a/logos_delivery/waku/rln/protocol_types.nim +++ b/logos_delivery/waku/rln/protocol_types.nim @@ -1,7 +1,7 @@ {.push raises: [].} import std/[tables, deques], stew/arrayops, stint, chronos, web3, eth/keys -import ../waku_core, ../waku_keystore, ../common/protobuf +import ../waku_core, ../waku_keystore, ../common/protobuf, ../common/protobuf_ext export waku_keystore, waku_core @@ -75,54 +75,49 @@ type MessageValidationResult* {.pure.} = enum Spam # Protobufs enc and init -proc init*(T: type RateLimitProof, buffer: seq[byte]): ProtoResult[T] = - var nsp: RateLimitProof +# externalNullifier not serialized +type RateLimitProofPB {.proto2.} = object + proof {.fieldNumber: 1, ext, required.}: ZKSNARK + merkleRoot {.fieldNumber: 2, ext, required.}: MerkleNode + epoch {.fieldNumber: 3, ext, required.}: Epoch + shareX {.fieldNumber: 4, ext, required.}: MerkleNode + shareY {.fieldNumber: 5, ext, required.}: MerkleNode + nullifier {.fieldNumber: 6, ext, required.}: Nullifier + rlnIdentifier {.fieldNumber: 7, ext, required.}: RlnIdentifier - let pb = initProtoBuffer(buffer) +proc decodeRateLimitProof(buffer: seq[byte]): ProtobufResult[RateLimitProof] = + var pb: RateLimitProofPB + try: + pb = Protobuf.decode(buffer, RateLimitProofPB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + ok( + RateLimitProof( + proof: pb.proof, + merkleRoot: pb.merkleRoot, + epoch: pb.epoch, + shareX: pb.shareX, + shareY: pb.shareY, + nullifier: pb.nullifier, + rlnIdentifier: pb.rlnIdentifier, + ) + ) - var proof: seq[byte] - discard ?pb.getField(1, proof) - discard nsp.proof.copyFrom(proof) +proc init*(T: type RateLimitProof, buffer: seq[byte]): ProtobufResult[T] = + decodeRateLimitProof(buffer) - var merkleRoot: seq[byte] - discard ?pb.getField(2, merkleRoot) - discard nsp.merkleRoot.copyFrom(merkleRoot) - - var epoch: seq[byte] - discard ?pb.getField(3, epoch) - discard nsp.epoch.copyFrom(epoch) - - var shareX: seq[byte] - discard ?pb.getField(4, shareX) - discard nsp.shareX.copyFrom(shareX) - - var shareY: seq[byte] - discard ?pb.getField(5, shareY) - discard nsp.shareY.copyFrom(shareY) - - var nullifier: seq[byte] - discard ?pb.getField(6, nullifier) - discard nsp.nullifier.copyFrom(nullifier) - - var rlnIdentifier: seq[byte] - discard ?pb.getField(7, rlnIdentifier) - discard nsp.rlnIdentifier.copyFrom(rlnIdentifier) - - return ok(nsp) - -proc encode*(nsp: RateLimitProof): ProtoBuffer = - var output = initProtoBuffer() - - output.write3(1, nsp.proof) - output.write3(2, nsp.merkleRoot) - output.write3(3, nsp.epoch) - output.write3(4, nsp.shareX) - output.write3(5, nsp.shareY) - output.write3(6, nsp.nullifier) - output.write3(7, nsp.rlnIdentifier) - - output.finish3() - return output +proc encode*(nsp: RateLimitProof): seq[byte] = + Protobuf.encode( + RateLimitProofPB( + proof: nsp.proof, + merkleRoot: nsp.merkleRoot, + epoch: nsp.epoch, + shareX: nsp.shareX, + shareY: nsp.shareY, + nullifier: nsp.nullifier, + rlnIdentifier: nsp.rlnIdentifier, + ) + ) func encode*(x: UInt32): seq[byte] = ## the Ethereum ABI imposes a 32 byte width for every type diff --git a/logos_delivery/waku/waku_core/message/codec.nim b/logos_delivery/waku/waku_core/message/codec.nim index 9b01cf667..9436fe52c 100644 --- a/logos_delivery/waku/waku_core/message/codec.nim +++ b/logos_delivery/waku/waku_core/message/codec.nim @@ -4,70 +4,93 @@ # - Proto definition: https://github.com/vacp2p/waku/blob/main/waku/message/v1/message.proto {.push raises: [].} -import ../../common/protobuf, ../topics, ../time, ./message +import ../../common/protobuf_ext +import ../../common/protobuf -proc encode*(message: WakuMessage): ProtoBuffer = - var buf = initProtoBuffer() +import ../time, ./message - buf.write3(1, message.payload) - buf.write3(2, message.contentTopic) - buf.write3(3, message.version) - buf.write3(10, zint64(message.timestamp)) - buf.write3(11, message.meta) - buf.write3(21, message.proof) - buf.write3(31, uint32(message.ephemeral)) - buf.finish3() +type WakuMessagePB {.proto2.} = object + payload {.fieldNumber: 1, required.}: seq[byte] + contentTopic {.fieldNumber: 2, required.}: string + version {.fieldNumber: 3, pint.}: Opt[uint32] + timestamp {.fieldNumber: 10, sint.}: Opt[int64] + meta {.fieldNumber: 11.}: Opt[seq[byte]] + proof {.fieldNumber: 21.}: Opt[seq[byte]] + ephemeral {.fieldNumber: 31.}: Opt[bool] - buf +proc encode*(message: WakuMessage): seq[byte] = + Protobuf.encode( + WakuMessagePB( + payload: message.payload, + contentTopic: message.contentTopic, + version: Opt.some(message.version), + timestamp: Opt.some(int64(message.timestamp)), + meta: Opt.some(message.meta), + proof: Opt.some(message.proof), + ephemeral: Opt.some(message.ephemeral), + ) + ) + +proc decodeWakuMessage(buffer: seq[byte]): ProtobufResult[WakuMessage] = + var pb: WakuMessagePB + try: + pb = Protobuf.decode(buffer, WakuMessagePB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + + let meta = pb.meta.get(@[]) + if meta.len > MaxMetaAttrLength: + return err(protobuf.ProtobufError.invalidLengthField("meta")) + + ok( + WakuMessage( + payload: pb.payload, + contentTopic: pb.contentTopic, + meta: meta, + version: pb.version.get(0'u32), + timestamp: Timestamp(pb.timestamp.get(0'i64)), + ephemeral: pb.ephemeral.get(false), + proof: pb.proof.get(@[]), + ) + ) proc decode*(T: type WakuMessage, buffer: seq[byte]): ProtobufResult[T] = - var msg = WakuMessage() - let pb = initProtoBuffer(buffer) + decodeWakuMessage(buffer) - var payload: seq[byte] - if not ?pb.getField(1, payload): - return err(ProtobufError.missingRequiredField("payload")) +# WakuMessage as a nested length-delimited field. +func supportsPacked*(T: type WakuMessage, ProtoType: type ProtobufExt): bool = + false + +func computeFieldSize*( + field: int, + value: WakuMessage, + ProtoType: type ProtobufExt, + skipDefault: static bool, +): int = + computeFieldSize(field, encode(value), pbytes, skipDefault) + +proc writeField*( + stream: OutputStream, + field: int, + value: WakuMessage, + ProtoType: type ProtobufExt, + skipDefault: static bool = false, +) {.raises: [IOError].} = + writeField(stream, field, encode(value), pbytes, skipDefault) + +proc readFieldInto*( + stream: InputStream, + value: var WakuMessage, + header: FieldHeader, + ProtoType: type ProtobufExt, +): bool {.raises: [SerializationError, IOError].} = + var s: seq[byte] + if readFieldInto(stream, s, header, pbytes): + let decoded = WakuMessage.decode(s) + if decoded.isOk(): + value = decoded.get() + true + else: + raise (ref ProtobufValueError)(msg: "Invalid nested WakuMessage") else: - msg.payload = payload - - var topic: ContentTopic - if not ?pb.getField(2, topic): - return err(ProtobufError.missingRequiredField("content_topic")) - else: - msg.contentTopic = topic - - var version: uint32 - if not ?pb.getField(3, version): - msg.version = 0 - else: - msg.version = version - - var timestamp: zint64 - if not ?pb.getField(10, timestamp): - msg.timestamp = Timestamp(0) - else: - msg.timestamp = Timestamp(timestamp) - - var meta: seq[byte] - if not ?pb.getField(11, meta): - msg.meta = @[] - else: - if meta.len > MaxMetaAttrLength: - return err(ProtobufError.invalidLengthField("meta")) - - msg.meta = meta - - # this is part of https://rfc.vac.dev/spec/17/ spec - var proof: seq[byte] - if not ?pb.getField(21, proof): - msg.proof = @[] - else: - msg.proof = proof - - var ephemeral: uint32 - if not ?pb.getField(31, ephemeral): - msg.ephemeral = false - else: - msg.ephemeral = bool(ephemeral) - - ok(msg) + false diff --git a/logos_delivery/waku/waku_filter_v2/client.nim b/logos_delivery/waku/waku_filter_v2/client.nim index ae12372ba..58e532f96 100644 --- a/logos_delivery/waku/waku_filter_v2/client.nim +++ b/logos_delivery/waku/waku_filter_v2/client.nim @@ -59,7 +59,7 @@ proc sendSubscribeRequest( await connection.closeWithEOF() try: - await connection.writeLP(filterSubscribeRequest.encode().buffer) + await connection.writeLP(filterSubscribeRequest.encode()) except CatchableError: let errMsg = "exception in waku_filter_v2 client writeLP: " & getCurrentExceptionMsg() diff --git a/logos_delivery/waku/waku_filter_v2/protocol.nim b/logos_delivery/waku/waku_filter_v2/protocol.nim index 768f303fe..ddada0278 100644 --- a/logos_delivery/waku/waku_filter_v2/protocol.nim +++ b/logos_delivery/waku/waku_filter_v2/protocol.nim @@ -210,7 +210,7 @@ proc pushToPeers( target_peer_ids = targetPeerIds, msg_hash = msgHash - let bufferToPublish = messagePush.encode().buffer + let bufferToPublish = messagePush.encode() var pushFuts: seq[Future[Result[void, string]]] for peerId in peers: @@ -335,7 +335,7 @@ proc initProtocolHandler(wf: WakuFilter) = ) try: - await conn.writeLp(response.encode().buffer) #TODO: toRPC() separation here + await conn.writeLp(response.encode()) #TODO: toRPC() separation here except LPStreamError: error "failed to write stream in writeLp", remote_peer_id = conn.peerId, error = getCurrentExceptionMsg() diff --git a/logos_delivery/waku/waku_filter_v2/rpc.nim b/logos_delivery/waku/waku_filter_v2/rpc.nim index 30c0742c0..a87526e2b 100644 --- a/logos_delivery/waku/waku_filter_v2/rpc.nim +++ b/logos_delivery/waku/waku_filter_v2/rpc.nim @@ -1,6 +1,11 @@ {.push raises: [].} -import results, json_serialization +import + results, + json_serialization, + protobuf_serialization, + protobuf_serialization/pkg/results, + protobuf_serialization/std/enums import ../waku_core type @@ -11,20 +16,22 @@ type UNSUBSCRIBE = uint32(2) UNSUBSCRIBE_ALL = uint32(3) - FilterSubscribeRequest* = object # Request from client to service node + FilterSubscribeRequest* = object + # Request from client to service node + # serialized via DTO in rpc_codec.nim; absent filterSubscribeType -> ping requestId*: string filterSubscribeType*: FilterSubscribeType pubsubTopic*: Opt[PubsubTopic] contentTopics*: seq[ContentTopic] - FilterSubscribeResponse* = object # Response from service node to client - requestId*: string - statusCode*: uint32 - statusDesc*: Opt[string] + FilterSubscribeResponse* {.proto2.} = object # Response from service node to client + requestId* {.fieldNumber: 1, required.}: string + statusCode* {.fieldNumber: 10, pint, required.}: uint32 + statusDesc* {.fieldNumber: 11.}: Opt[string] - MessagePush* = object # Message pushed from service node to client - wakuMessage*: WakuMessage - pubsubTopic*: string + MessagePush* {.proto2.} = object # Message pushed from service node to client + wakuMessage* {.fieldNumber: 1, ext, required.}: WakuMessage + pubsubTopic* {.fieldNumber: 2, required.}: string # Convenience functions diff --git a/logos_delivery/waku/waku_filter_v2/rpc_codec.nim b/logos_delivery/waku/waku_filter_v2/rpc_codec.nim index 98c7111e2..d27a852c4 100644 --- a/logos_delivery/waku/waku_filter_v2/rpc_codec.nim +++ b/logos_delivery/waku/waku_filter_v2/rpc_codec.nim @@ -1,6 +1,10 @@ {.push raises: [].} -import results, ../common/protobuf, ../waku_core, ./rpc +import + protobuf_serialization, + protobuf_serialization/pkg/results, + protobuf_serialization/std/enums +import ../common/protobuf, ../waku_core, ./rpc const DefaultMaxSubscribeSize* = 10 * DefaultMaxWakuMessageSize + 64 * 1024 @@ -9,89 +13,67 @@ const DefaultMaxPushSize* = 10 * DefaultMaxWakuMessageSize + 64 * 1024 # We add a 64kB safety buffer for protocol overhead -proc encode*(rpc: FilterSubscribeRequest): ProtoBuffer = - var pb = initProtoBuffer() +# absent field 2 (Opt) decodes to SUBSCRIBER_PING +type FilterSubscribeRequestPB {.proto2.} = object + requestId {.fieldNumber: 1, required.}: string + filterSubscribeType {.fieldNumber: 2, ext.}: Opt[FilterSubscribeType] + pubsubTopic {.fieldNumber: 10.}: Opt[PubsubTopic] + contentTopics {.fieldNumber: 11.}: seq[ContentTopic] - pb.write3(1, rpc.requestId) - pb.write3(2, uint32(ord(rpc.filterSubscribeType))) +proc encode*(rpc: FilterSubscribeRequest): seq[byte] = + Protobuf.encode( + FilterSubscribeRequestPB( + requestId: rpc.requestId, + filterSubscribeType: Opt.some(rpc.filterSubscribeType), + pubsubTopic: rpc.pubsubTopic, + contentTopics: rpc.contentTopics, + ) + ) - pb.write3(10, rpc.pubsubTopic) +proc encode*(rpc: FilterSubscribeResponse): seq[byte] = + Protobuf.encode(rpc) - for contentTopic in rpc.contentTopics: - pb.write3(11, contentTopic) +proc encode*(rpc: MessagePush): seq[byte] = + Protobuf.encode(rpc) - pb +# non-generic: mixin Reader must resolve here +proc decodeFilterSubscribeRequest( + buffer: seq[byte] +): ProtobufResult[FilterSubscribeRequest] = + var pb: FilterSubscribeRequestPB + try: + pb = Protobuf.decode(buffer, FilterSubscribeRequestPB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + ok( + FilterSubscribeRequest( + requestId: pb.requestId, + filterSubscribeType: + pb.filterSubscribeType.get(FilterSubscribeType.SUBSCRIBER_PING), + pubsubTopic: pb.pubsubTopic, + contentTopics: pb.contentTopics, + ) + ) + +proc decodeFilterSubscribeResponse( + buffer: seq[byte] +): ProtobufResult[FilterSubscribeResponse] = + try: + ok(Protobuf.decode(buffer, FilterSubscribeResponse)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + +proc decodeMessagePush(buffer: seq[byte]): ProtobufResult[MessagePush] = + try: + ok(Protobuf.decode(buffer, MessagePush)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) proc decode*(T: type FilterSubscribeRequest, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = FilterSubscribeRequest() - - if not ?pb.getField(1, rpc.requestId): - return err(ProtobufError.missingRequiredField("request_id")) - - var filterSubscribeType: uint32 - if not ?pb.getField(2, filterSubscribeType): - # Revert to ping by default if not set - rpc.filterSubscribeType = FilterSubscribeType.SUBSCRIBER_PING - else: - rpc.filterSubscribeType = FilterSubscribeType(filterSubscribeType) - - var pubsubTopic: PubsubTopic - if not ?pb.getField(10, pubsubTopic): - rpc.pubsubTopic = Opt.none(PubsubTopic) - else: - rpc.pubsubTopic = Opt.some(pubsubTopic) - - discard ?pb.getRepeatedField(11, rpc.contentTopics) - - ok(rpc) - -proc encode*(rpc: FilterSubscribeResponse): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, rpc.requestId) - pb.write3(10, rpc.statusCode) - pb.write3(11, rpc.statusDesc) - - pb + decodeFilterSubscribeRequest(buffer) proc decode*(T: type FilterSubscribeResponse, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = FilterSubscribeResponse() - - if not ?pb.getField(1, rpc.requestId): - return err(ProtobufError.missingRequiredField("request_id")) - - if not ?pb.getField(10, rpc.statusCode): - return err(ProtobufError.missingRequiredField("status_code")) - - var statusDesc: string - if not ?pb.getField(11, statusDesc): - rpc.statusDesc = Opt.none(string) - else: - rpc.statusDesc = Opt.some(statusDesc) - - ok(rpc) - -proc encode*(rpc: MessagePush): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, rpc.wakuMessage.encode()) - pb.write3(2, rpc.pubsubTopic) - - pb + decodeFilterSubscribeResponse(buffer) proc decode*(T: type MessagePush, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = MessagePush() - - var message: seq[byte] - if not ?pb.getField(1, message): - return err(ProtobufError.missingRequiredField("message")) - else: - rpc.wakuMessage = ?WakuMessage.decode(message) - - if not ?pb.getField(2, rpc.pubsubTopic): - return err(ProtobufError.missingRequiredField("pubsub_topic")) - - ok(rpc) + decodeMessagePush(buffer) diff --git a/logos_delivery/waku/waku_lightpush/client.nim b/logos_delivery/waku/waku_lightpush/client.nim index b282c2fa3..6e21e77f6 100644 --- a/logos_delivery/waku/waku_lightpush/client.nim +++ b/logos_delivery/waku/waku_lightpush/client.nim @@ -47,7 +47,7 @@ proc sendPushRequest( defer: await connection.closeWithEOF() - await connection.writeLP(req.encode().buffer) + await connection.writeLP(req.encode()) var buffer: seq[byte] try: diff --git a/logos_delivery/waku/waku_lightpush/protocol.nim b/logos_delivery/waku/waku_lightpush/protocol.nim index a698dd344..d67398956 100644 --- a/logos_delivery/waku/waku_lightpush/protocol.nim +++ b/logos_delivery/waku/waku_lightpush/protocol.nim @@ -138,7 +138,7 @@ proc initProtocolHandler(wl: WakuLightPush) = ) try: - await conn.writeLp(rpc.encode().buffer) + await conn.writeLp(rpc.encode()) except LPStreamError: error "lightpush write stream failed", error = getCurrentExceptionMsg() diff --git a/logos_delivery/waku/waku_lightpush/rpc.nim b/logos_delivery/waku/waku_lightpush/rpc.nim index 228d9f836..177fe8d03 100644 --- a/logos_delivery/waku/waku_lightpush/rpc.nim +++ b/logos_delivery/waku/waku_lightpush/rpc.nim @@ -1,19 +1,54 @@ {.push raises: [].} -import results, ../waku_core +import results, protobuf_serialization, protobuf_serialization/pkg/results +import ../waku_core type LightPushStatusCode* = distinct uint32 proc `==`*(a, b: LightPushStatusCode): bool {.borrow.} proc `$`*(code: LightPushStatusCode): string {.borrow.} -type - LightpushRequest* = object - requestId*: string - pubSubTopic*: Opt[PubsubTopic] - message*: WakuMessage +# LightPushStatusCode (distinct uint32) as a plain varint +func supportsPacked*(T: type LightPushStatusCode, ProtoType: type ProtobufExt): bool = + false - LightPushResponse* = object - requestId*: string - statusCode*: LightPushStatusCode - statusDesc*: Opt[string] - relayPeerCount*: Opt[uint32] +func computeFieldSize*( + field: int, + value: LightPushStatusCode, + ProtoType: type ProtobufExt, + skipDefault: static bool, +): int = + computeFieldSize(field, uint32(value), puint32, skipDefault) + +proc writeField*( + stream: OutputStream, + field: int, + value: LightPushStatusCode, + ProtoType: type ProtobufExt, + skipDefault: static bool = false, +) {.raises: [IOError].} = + writeField(stream, field, uint32(value), puint32, skipDefault) + +proc readFieldInto*( + stream: InputStream, + value: var LightPushStatusCode, + header: FieldHeader, + ProtoType: type ProtobufExt, +): bool {.raises: [SerializationError, IOError].} = + var v: uint32 + if readFieldInto(stream, v, header, puint32): + value = LightPushStatusCode(v) + true + else: + false + +type + LightpushRequest* {.proto2.} = object + requestId* {.fieldNumber: 1, required.}: string + pubSubTopic* {.fieldNumber: 20.}: Opt[PubsubTopic] + message* {.fieldNumber: 21, ext, required.}: WakuMessage + + LightPushResponse* {.proto2.} = object + requestId* {.fieldNumber: 1, required.}: string + statusCode* {.fieldNumber: 10, ext, required.}: LightPushStatusCode + statusDesc* {.fieldNumber: 11.}: Opt[string] + relayPeerCount* {.fieldNumber: 12, pint.}: Opt[uint32] diff --git a/logos_delivery/waku/waku_lightpush/rpc_codec.nim b/logos_delivery/waku/waku_lightpush/rpc_codec.nim index cec295f24..ed356e5be 100644 --- a/logos_delivery/waku/waku_lightpush/rpc_codec.nim +++ b/logos_delivery/waku/waku_lightpush/rpc_codec.nim @@ -1,80 +1,30 @@ {.push raises: [].} -import results, ../common/protobuf, ../waku_core, ./rpc +import protobuf_serialization, protobuf_serialization/pkg/results +import ../common/protobuf, ../waku_core, ./rpc const DefaultMaxRpcSize* = -1 -proc encode*(rpc: LightpushRequest): ProtoBuffer = - var pb = initProtoBuffer() +proc encode*(rpc: LightpushRequest): seq[byte] = + Protobuf.encode(rpc) - pb.write3(1, rpc.requestId) - pb.write3(20, rpc.pubSubTopic) - pb.write3(21, rpc.message.encode()) - pb.finish3() +proc encode*(rpc: LightPushResponse): seq[byte] = + Protobuf.encode(rpc) - return pb +proc decodeLightpushRequest(buffer: seq[byte]): ProtobufResult[LightpushRequest] = + try: + ok(Protobuf.decode(buffer, LightpushRequest)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + +proc decodeLightPushResponse(buffer: seq[byte]): ProtobufResult[LightPushResponse] = + try: + ok(Protobuf.decode(buffer, LightPushResponse)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) proc decode*(T: type LightpushRequest, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = LightpushRequest() - - var requestId: string - if not ?pb.getField(1, requestId): - return err(ProtobufError.missingRequiredField("request_id")) - else: - rpc.requestId = requestId - - var pubSubTopic: PubsubTopic - if not ?pb.getField(20, pubSubTopic): - rpc.pubSubTopic = Opt.none(PubsubTopic) - else: - rpc.pubSubTopic = Opt.some(pubSubTopic) - - var messageBuf: seq[byte] - if not ?pb.getField(21, messageBuf): - return err(ProtobufError.missingRequiredField("message")) - else: - rpc.message = ?WakuMessage.decode(messageBuf) - - return ok(rpc) - -proc encode*(rpc: LightPushResponse): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, rpc.requestId) - pb.write3(10, rpc.statusCode.uint32) - pb.write3(11, rpc.statusDesc) - pb.write3(12, rpc.relayPeerCount) - pb.finish3() - - return pb + decodeLightpushRequest(buffer) proc decode*(T: type LightPushResponse, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = LightPushResponse() - - var requestId: string - if not ?pb.getField(1, requestId): - return err(ProtobufError.missingRequiredField("request_id")) - else: - rpc.requestId = requestId - - var statusCode: uint32 - if not ?pb.getField(10, statusCode): - return err(ProtobufError.missingRequiredField("status_code")) - else: - rpc.statusCode = statusCode.LightPushStatusCode - - var statusDesc: string - if not ?pb.getField(11, statusDesc): - rpc.statusDesc = Opt.none(string) - else: - rpc.statusDesc = Opt.some(statusDesc) - - var relayPeerCount: uint32 - if not ?pb.getField(12, relayPeerCount): - rpc.relayPeerCount = Opt.none(uint32) - else: - rpc.relayPeerCount = Opt.some(relayPeerCount) - - return ok(rpc) + decodeLightPushResponse(buffer) diff --git a/logos_delivery/waku/waku_lightpush/self_req_handler.nim b/logos_delivery/waku/waku_lightpush/self_req_handler.nim index 141af9ad9..86593da9e 100644 --- a/logos_delivery/waku/waku_lightpush/self_req_handler.nim +++ b/logos_delivery/waku/waku_lightpush/self_req_handler.nim @@ -27,7 +27,7 @@ proc handleSelfLightPushRequest*( requestId: generateRequestId(self.rng), pubSubTopic: pubSubTopic, message: message ) - let response = await self.handleRequest(selfPeerId, req.encode().buffer) + let response = await self.handleRequest(selfPeerId, req.encode()) return response.toPushResult() except Exception: diff --git a/logos_delivery/waku/waku_lightpush_legacy/client.nim b/logos_delivery/waku/waku_lightpush_legacy/client.nim index 1cd3a4302..cc1c3c4a8 100644 --- a/logos_delivery/waku/waku_lightpush_legacy/client.nim +++ b/logos_delivery/waku/waku_lightpush_legacy/client.nim @@ -38,7 +38,7 @@ proc sendPushRequest( await connection.closeWithEOF() let rpc = PushRPC(requestId: generateRequestId(wl.rng), request: Opt.some(req)) - await connection.writeLP(rpc.encode().buffer) + await connection.writeLP(rpc.encode()) var buffer: seq[byte] try: diff --git a/logos_delivery/waku/waku_lightpush_legacy/protocol.nim b/logos_delivery/waku/waku_lightpush_legacy/protocol.nim index ba982f8f1..60b393709 100644 --- a/logos_delivery/waku/waku_lightpush_legacy/protocol.nim +++ b/logos_delivery/waku/waku_lightpush_legacy/protocol.nim @@ -105,7 +105,7 @@ proc initProtocolHandler(wl: WakuLegacyLightPush) = ) try: - await conn.writeLp(rpc.encode().buffer) + await conn.writeLp(rpc.encode()) except LPStreamError: error "lightpush legacy write stream failed", error = getCurrentExceptionMsg() diff --git a/logos_delivery/waku/waku_lightpush_legacy/rpc.nim b/logos_delivery/waku/waku_lightpush_legacy/rpc.nim index ba92bd55f..1d6a651b5 100644 --- a/logos_delivery/waku/waku_lightpush_legacy/rpc.nim +++ b/logos_delivery/waku/waku_lightpush_legacy/rpc.nim @@ -1,17 +1,17 @@ {.push raises: [].} -import results, ../waku_core +import results, protobuf_serialization, protobuf_serialization/pkg/results, ../waku_core type - PushRequest* = object - pubSubTopic*: string - message*: WakuMessage + PushRequest* {.proto2.} = object + pubSubTopic* {.fieldNumber: 1, required.}: string + message* {.fieldNumber: 2, ext, required.}: WakuMessage - PushResponse* = object - isSuccess*: bool - info*: Opt[string] + PushResponse* {.proto2.} = object + isSuccess* {.fieldNumber: 1, required.}: bool + info* {.fieldNumber: 2.}: Opt[string] - PushRPC* = object - requestId*: string - request*: Opt[PushRequest] - response*: Opt[PushResponse] + PushRPC* {.proto2.} = object + requestId* {.fieldNumber: 1, required.}: string + request* {.fieldNumber: 2.}: Opt[PushRequest] + response* {.fieldNumber: 3.}: Opt[PushResponse] diff --git a/logos_delivery/waku/waku_lightpush_legacy/rpc_codec.nim b/logos_delivery/waku/waku_lightpush_legacy/rpc_codec.nim index a31c71f03..f90eeab12 100644 --- a/logos_delivery/waku/waku_lightpush_legacy/rpc_codec.nim +++ b/logos_delivery/waku/waku_lightpush_legacy/rpc_codec.nim @@ -1,95 +1,43 @@ {.push raises: [].} -import results, ../common/protobuf, ../waku_core, ./rpc +import protobuf_serialization, protobuf_serialization/pkg/results +import ../common/protobuf, ../waku_core, ./rpc const DefaultMaxRpcSize* = -1 -proc encode*(rpc: PushRequest): ProtoBuffer = - var pb = initProtoBuffer() +proc encode*(rpc: PushRequest): seq[byte] = + Protobuf.encode(rpc) - pb.write3(1, rpc.pubSubTopic) - pb.write3(2, rpc.message.encode()) - pb.finish3() +proc encode*(rpc: PushResponse): seq[byte] = + Protobuf.encode(rpc) - pb +proc encode*(rpc: PushRPC): seq[byte] = + Protobuf.encode(rpc) + +# non-generic: mixin Reader must resolve here +proc decodePushRequest(buffer: seq[byte]): ProtobufResult[PushRequest] = + try: + ok(Protobuf.decode(buffer, PushRequest)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + +proc decodePushResponse(buffer: seq[byte]): ProtobufResult[PushResponse] = + try: + ok(Protobuf.decode(buffer, PushResponse)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + +proc decodePushRPC(buffer: seq[byte]): ProtobufResult[PushRPC] = + try: + ok(Protobuf.decode(buffer, PushRPC)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) proc decode*(T: type PushRequest, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = PushRequest() - - var pubSubTopic: PubsubTopic - if not ?pb.getField(1, pubSubTopic): - return err(ProtobufError.missingRequiredField("pubsub_topic")) - else: - rpc.pubSubTopic = pubSubTopic - - var messageBuf: seq[byte] - if not ?pb.getField(2, messageBuf): - return err(ProtobufError.missingRequiredField("message")) - else: - rpc.message = ?WakuMessage.decode(messageBuf) - - ok(rpc) - -proc encode*(rpc: PushResponse): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, uint64(rpc.isSuccess)) - pb.write3(2, rpc.info) - pb.finish3() - - pb + decodePushRequest(buffer) proc decode*(T: type PushResponse, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = PushResponse() - - var isSuccess: uint64 - if not ?pb.getField(1, isSuccess): - return err(ProtobufError.missingRequiredField("is_success")) - else: - rpc.isSuccess = bool(isSuccess) - - var info: string - if not ?pb.getField(2, info): - rpc.info = Opt.none(string) - else: - rpc.info = Opt.some(info) - - ok(rpc) - -proc encode*(rpc: PushRPC): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, rpc.requestId) - pb.write3(2, rpc.request.map(encode)) - pb.write3(3, rpc.response.map(encode)) - pb.finish3() - - pb + decodePushResponse(buffer) proc decode*(T: type PushRPC, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = PushRPC() - - var requestId: string - if not ?pb.getField(1, requestId): - return err(ProtobufError.missingRequiredField("request_id")) - else: - rpc.requestId = requestId - - var requestBuffer: seq[byte] - if not ?pb.getField(2, requestBuffer): - rpc.request = Opt.none(PushRequest) - else: - let request = ?PushRequest.decode(requestBuffer) - rpc.request = Opt.some(request) - - var responseBuffer: seq[byte] - if not ?pb.getField(3, responseBuffer): - rpc.response = Opt.none(PushResponse) - else: - let response = ?PushResponse.decode(responseBuffer) - rpc.response = Opt.some(response) - - ok(rpc) + decodePushRPC(buffer) diff --git a/logos_delivery/waku/waku_lightpush_legacy/self_req_handler.nim b/logos_delivery/waku/waku_lightpush_legacy/self_req_handler.nim index c1c04ad44..59d4d9e87 100644 --- a/logos_delivery/waku/waku_lightpush_legacy/self_req_handler.nim +++ b/logos_delivery/waku/waku_lightpush_legacy/self_req_handler.nim @@ -33,7 +33,7 @@ proc handleSelfLightPushRequest*( let req = PushRequest(pubSubTopic: pubSubTopic, message: message) let rpc = PushRPC(requestId: generateRequestId(self.rng), request: Opt.some(req)) - let respRpc = await self.handleRequest(selfPeerId, rpc.encode().buffer) + let respRpc = await self.handleRequest(selfPeerId, rpc.encode()) if respRpc.response.isNone(): waku_lightpush_errors.inc(labelValues = [emptyResponseBodyFailure]) diff --git a/logos_delivery/waku/waku_metadata/protocol.nim b/logos_delivery/waku/waku_metadata/protocol.nim index 2a59fd487..a051947ae 100644 --- a/logos_delivery/waku/waku_metadata/protocol.nim +++ b/logos_delivery/waku/waku_metadata/protocol.nim @@ -32,7 +32,7 @@ proc respond( ) let res = catch: - await conn.writeLP(response.encode().buffer) + await conn.writeLP(response.encode()) res.isOkOr: return err(error.msg) @@ -46,7 +46,7 @@ proc request*( ) let writeRes = catch: - await conn.writeLP(request.encode().buffer) + await conn.writeLP(request.encode()) let readRes = catch: await conn.readLp(RpcResponseMaxBytes) diff --git a/logos_delivery/waku/waku_metadata/rpc.nim b/logos_delivery/waku/waku_metadata/rpc.nim index 37a9e77d0..1936aab6c 100644 --- a/logos_delivery/waku/waku_metadata/rpc.nim +++ b/logos_delivery/waku/waku_metadata/rpc.nim @@ -1,7 +1,7 @@ {.push raises: [].} import results - +import protobuf_serialization, protobuf_serialization/pkg/results import ../common/protobuf type WakuMetadataRequest* = object @@ -12,71 +12,34 @@ type WakuMetadataResponse* = object clusterId*: Opt[uint32] shards*: seq[uint32] -proc encode*(rpc: WakuMetadataRequest): ProtoBuffer = - var pb = initProtoBuffer() +# shards emitted twice: unpacked field 2 (deprecated) + packed field 3 +type WakuMetadataPB {.proto2.} = object + clusterId {.fieldNumber: 1, pint.}: Opt[uint32] + shardsDeprecated {.fieldNumber: 2, pint, packed: false.}: seq[uint32] + shardsPacked {.fieldNumber: 3, pint, packed: true.}: seq[uint32] - pb.write3(1, rpc.clusterId) - for shard in rpc.shards: - pb.write3(2, shard) # deprecated - pb.writePacked(3, rpc.shards) - pb.finish3() +proc toPB(clusterId: Opt[uint32], shards: seq[uint32]): WakuMetadataPB = + WakuMetadataPB(clusterId: clusterId, shardsDeprecated: shards, shardsPacked: shards) - pb +proc shardsFrom(pb: WakuMetadataPB): seq[uint32] = + if pb.shardsPacked.len > 0: pb.shardsPacked else: pb.shardsDeprecated -proc decode*(T: type WakuMetadataRequest, buffer: seq[byte]): ProtoResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = WakuMetadataRequest() +proc decodeMetadataPB(buffer: seq[byte]): ProtobufResult[WakuMetadataPB] = + try: + ok(Protobuf.decode(buffer, WakuMetadataPB)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) - var clusterId: uint64 - if not ?pb.getField(1, clusterId): - rpc.clusterId = Opt.none(uint32) - else: - rpc.clusterId = Opt.some(clusterId.uint32) +proc encode*(rpc: WakuMetadataRequest): seq[byte] = + Protobuf.encode(toPB(rpc.clusterId, rpc.shards)) - var shards: seq[uint64] - if ?pb.getPackedRepeatedField(3, shards): - for shard in shards: - rpc.shards.add(shard.uint32) - elif ?pb.getPackedRepeatedField(2, shards): - for shard in shards: - rpc.shards.add(shard.uint32) - elif ?pb.getRepeatedField(2, shards): - for shard in shards: - rpc.shards.add(shard.uint32) +proc decode*(T: type WakuMetadataRequest, buffer: seq[byte]): ProtobufResult[T] = + let pb = ?decodeMetadataPB(buffer) + ok(WakuMetadataRequest(clusterId: pb.clusterId, shards: shardsFrom(pb))) - ok(rpc) +proc encode*(rpc: WakuMetadataResponse): seq[byte] = + Protobuf.encode(toPB(rpc.clusterId, rpc.shards)) -proc encode*(rpc: WakuMetadataResponse): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, rpc.clusterId) - for shard in rpc.shards: - pb.write3(2, shard) # deprecated - pb.writePacked(3, rpc.shards) - pb.finish3() - - pb - -proc decode*(T: type WakuMetadataResponse, buffer: seq[byte]): ProtoResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = WakuMetadataResponse() - - var clusterId: uint64 - if not ?pb.getField(1, clusterId): - rpc.clusterId = Opt.none(uint32) - else: - rpc.clusterId = Opt.some(clusterId.uint32) - - var shards: seq[uint64] - - if ?pb.getPackedRepeatedField(3, shards): - for shard in shards: - rpc.shards.add(shard.uint32) - elif ?pb.getPackedRepeatedField(2, shards): - for shard in shards: - rpc.shards.add(shard.uint32) - elif ?pb.getRepeatedField(2, shards): - for shard in shards: - rpc.shards.add(shard.uint32) - - ok(rpc) +proc decode*(T: type WakuMetadataResponse, buffer: seq[byte]): ProtobufResult[T] = + let pb = ?decodeMetadataPB(buffer) + ok(WakuMetadataResponse(clusterId: pb.clusterId, shards: shardsFrom(pb))) diff --git a/logos_delivery/waku/waku_peer_exchange/client.nim b/logos_delivery/waku/waku_peer_exchange/client.nim index f2ceb67c4..b312d3d25 100644 --- a/logos_delivery/waku/waku_peer_exchange/client.nim +++ b/logos_delivery/waku/waku_peer_exchange/client.nim @@ -33,7 +33,7 @@ proc request*( var callResult = (status_code: PeerExchangeResponseStatusCode.SUCCESS, status_desc: Opt.none(string)) try: - await conn.writeLP(rpc.encode().buffer) + await conn.writeLP(rpc.encode()) buffer = await conn.readLp(DefaultMaxRpcSize.int) except CatchableError as exc: error "exception when handling peer exchange request", error = exc.msg diff --git a/logos_delivery/waku/waku_peer_exchange/protocol.nim b/logos_delivery/waku/waku_peer_exchange/protocol.nim index 1e76be230..4d208a687 100644 --- a/logos_delivery/waku/waku_peer_exchange/protocol.nim +++ b/logos_delivery/waku/waku_peer_exchange/protocol.nim @@ -42,7 +42,7 @@ proc respond( let rpc = PeerExchangeRpc.makeResponse(enrs.mapIt(PeerExchangePeerInfo(enr: it.raw))) try: - await conn.writeLP(rpc.encode().buffer) + await conn.writeLP(rpc.encode()) except LPStreamError as exc: # Remote closed the stream before we responded - expected during peer churn. debug "peer exchange response not delivered: stream closed", @@ -66,7 +66,7 @@ proc respondError( let rpc = PeerExchangeRpc.makeErrorResponse(status_code, status_desc) try: - await conn.writeLP(rpc.encode().buffer) + await conn.writeLP(rpc.encode()) except LPStreamError as exc: # Remote closed the stream before we responded - expected during peer churn. debug "peer exchange error response not delivered: stream closed", diff --git a/logos_delivery/waku/waku_peer_exchange/rpc_codec.nim b/logos_delivery/waku/waku_peer_exchange/rpc_codec.nim index 00aefd275..4e19c31c8 100644 --- a/logos_delivery/waku/waku_peer_exchange/rpc_codec.nim +++ b/logos_delivery/waku/waku_peer_exchange/rpc_codec.nim @@ -1,42 +1,24 @@ {.push raises: [].} -import results, ../common/protobuf, ./rpc +import std/sequtils, results +import protobuf_serialization, protobuf_serialization/pkg/results +import ../common/protobuf, ./rpc -proc encode*(rpc: PeerExchangeRequest): ProtoBuffer = - var pb = initProtoBuffer() +type + PeerExchangePeerInfoPB {.proto2.} = object + enr {.fieldNumber: 1.}: Opt[seq[byte]] - pb.write3(1, rpc.numPeers) - pb.finish3() + PeerExchangeRequestPB {.proto2.} = object + numPeers {.fieldNumber: 1, pint.}: Opt[uint64] - pb + PeerExchangeResponsePB {.proto2.} = object + peerInfos {.fieldNumber: 1.}: seq[PeerExchangePeerInfoPB] + statusCode {.fieldNumber: 10, pint.}: Opt[uint32] + statusDesc {.fieldNumber: 11.}: Opt[string] -proc decode*(T: type PeerExchangeRequest, buffer: seq[byte]): ProtoResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = PeerExchangeRequest(numPeers: 0) - - var numPeers: uint64 - if ?pb.getField(1, numPeers): - rpc.numPeers = numPeers - - ok(rpc) - -proc encode*(rpc: PeerExchangePeerInfo): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, rpc.enr) - pb.finish3() - - pb - -proc decode*(T: type PeerExchangePeerInfo, buffer: seq[byte]): ProtoResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = PeerExchangePeerInfo(enr: @[]) - - var peerInfoBuffer: seq[byte] - if ?pb.getField(1, peerInfoBuffer): - rpc.enr = peerInfoBuffer - - ok(rpc) + PeerExchangeRpcPB {.proto2.} = object + request {.fieldNumber: 1, required.}: PeerExchangeRequestPB + response {.fieldNumber: 2.}: Opt[PeerExchangeResponsePB] proc parse*(T: type PeerExchangeResponseStatusCode, status: uint32): T = case status @@ -45,70 +27,98 @@ proc parse*(T: type PeerExchangeResponseStatusCode, status: uint32): T = else: PeerExchangeResponseStatusCode.UNKNOWN -proc encode*(rpc: PeerExchangeResponse): ProtoBuffer = - var pb = initProtoBuffer() +proc toPB(pi: PeerExchangePeerInfo): PeerExchangePeerInfoPB = + PeerExchangePeerInfoPB(enr: Opt.some(pi.enr)) - for pi in rpc.peerInfos: - pb.write3(1, pi.encode()) - pb.write3(10, rpc.status_code.uint32) - pb.write3(11, rpc.status_desc) +proc fromPB(pb: PeerExchangePeerInfoPB): PeerExchangePeerInfo = + PeerExchangePeerInfo(enr: pb.enr.get(@[])) - pb.finish3() +proc toPB(req: PeerExchangeRequest): PeerExchangeRequestPB = + PeerExchangeRequestPB(numPeers: Opt.some(req.numPeers)) - pb +proc fromPB(pb: PeerExchangeRequestPB): PeerExchangeRequest = + PeerExchangeRequest(numPeers: pb.numPeers.get(0'u64)) + +proc toPB(res: PeerExchangeResponse): PeerExchangeResponsePB = + PeerExchangeResponsePB( + peerInfos: res.peerInfos.mapIt(toPB(it)), + statusCode: Opt.some(uint32(ord(res.status_code))), + statusDesc: res.status_desc, + ) + +proc fromPB(pb: PeerExchangeResponsePB): PeerExchangeResponse = + let peerInfos = pb.peerInfos.mapIt(fromPB(it)) + let statusCode = + if pb.statusCode.isSome(): + PeerExchangeResponseStatusCode.parse(pb.statusCode.get()) + elif peerInfos.len() > 0: + # older peers may not support the status_code field yet + PeerExchangeResponseStatusCode.SUCCESS + else: + PeerExchangeResponseStatusCode.SERVICE_UNAVAILABLE + PeerExchangeResponse( + peerInfos: peerInfos, status_code: statusCode, status_desc: pb.statusDesc + ) + +proc encode*(rpc: PeerExchangeRequest): seq[byte] = + Protobuf.encode(toPB(rpc)) + +proc encode*(rpc: PeerExchangePeerInfo): seq[byte] = + Protobuf.encode(toPB(rpc)) + +proc encode*(rpc: PeerExchangeResponse): seq[byte] = + Protobuf.encode(toPB(rpc)) + +proc encode*(rpc: PeerExchangeRpc): seq[byte] = + Protobuf.encode( + PeerExchangeRpcPB( + request: toPB(rpc.request), response: Opt.some(toPB(rpc.response)) + ) + ) + +proc decodePeerExchangeRequest(buffer: seq[byte]): ProtobufResult[PeerExchangeRequest] = + try: + ok(fromPB(Protobuf.decode(buffer, PeerExchangeRequestPB))) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + +proc decodePeerExchangePeerInfo( + buffer: seq[byte] +): ProtobufResult[PeerExchangePeerInfo] = + try: + ok(fromPB(Protobuf.decode(buffer, PeerExchangePeerInfoPB))) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + +proc decodePeerExchangeResponse( + buffer: seq[byte] +): ProtobufResult[PeerExchangeResponse] = + try: + ok(fromPB(Protobuf.decode(buffer, PeerExchangeResponsePB))) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + +proc decodePeerExchangeRpc(buffer: seq[byte]): ProtobufResult[PeerExchangeRpc] = + var pb: PeerExchangeRpcPB + try: + pb = Protobuf.decode(buffer, PeerExchangeRpcPB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + let response = + if pb.response.isSome(): + fromPB(pb.response.get()) + else: + PeerExchangeResponse(status_code: PeerExchangeResponseStatusCode.UNKNOWN) + ok(PeerExchangeRpc(request: fromPB(pb.request), response: response)) + +proc decode*(T: type PeerExchangeRequest, buffer: seq[byte]): ProtobufResult[T] = + decodePeerExchangeRequest(buffer) + +proc decode*(T: type PeerExchangePeerInfo, buffer: seq[byte]): ProtobufResult[T] = + decodePeerExchangePeerInfo(buffer) proc decode*(T: type PeerExchangeResponse, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = PeerExchangeResponse(peerInfos: @[]) - - var peerInfoBuffers: seq[seq[byte]] - if ?pb.getRepeatedField(1, peerInfoBuffers): - for pib in peerInfoBuffers: - rpc.peerInfos.add(?PeerExchangePeerInfo.decode(pib)) - - var status_code: uint32 - if ?pb.getField(10, status_code): - rpc.status_code = PeerExchangeResponseStatusCode.parse(status_code) - else: - # older peers may not support status_code field yet - if rpc.peerInfos.len() > 0: - rpc.status_code = PeerExchangeResponseStatusCode.SUCCESS - else: - rpc.status_code = PeerExchangeResponseStatusCode.SERVICE_UNAVAILABLE - - var status_desc: string - if ?pb.getField(11, status_desc): - rpc.status_desc = Opt.some(status_desc) - else: - rpc.status_desc = Opt.none(string) - - ok(rpc) - -proc encode*(rpc: PeerExchangeRpc): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, rpc.request.encode()) - pb.write3(2, rpc.response.encode()) - - pb.finish3() - - pb + decodePeerExchangeResponse(buffer) proc decode*(T: type PeerExchangeRpc, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - var rpc = PeerExchangeRpc() - - var requestBuffer: seq[byte] - if not ?pb.getField(1, requestBuffer): - return err(ProtobufError.missingRequiredField("request")) - - rpc.request = ?PeerExchangeRequest.decode(requestBuffer) - - var responseBuffer: seq[byte] - if not ?pb.getField(2, responseBuffer): - rpc.response = - PeerExchangeResponse(status_code: PeerExchangeResponseStatusCode.UNKNOWN) - else: - rpc.response = ?PeerExchangeResponse.decode(responseBuffer) - - ok(rpc) + decodePeerExchangeRpc(buffer) diff --git a/logos_delivery/waku/waku_relay/protocol.nim b/logos_delivery/waku/waku_relay/protocol.nim index 0997a1fd8..4eddc6823 100644 --- a/logos_delivery/waku/waku_relay/protocol.nim +++ b/logos_delivery/waku/waku_relay/protocol.nim @@ -566,7 +566,7 @@ proc generateOrderedValidator(w: WakuRelay): ValidatorHandler {.gcsafe.} = proc validateMessage*( w: WakuRelay, pubsubTopic: string, msg: WakuMessage ): Future[Result[void, string]] {.async.} = - let messageSizeBytes = msg.encode().buffer.len + let messageSizeBytes = msg.encode().len let msgHash = computeMessageHash(pubsubTopic, msg).to0xHex() if messageSizeBytes > w.maxMessageSize: @@ -680,7 +680,7 @@ proc publish*( if message.timestamp == 0: message.timestamp = getNowInNanosecondTime() - let data = message.encode().buffer + let data = message.encode() let msgHash = computeMessageHash(pubsubTopic, message).to0xHex() notice "start publish Waku message", diff --git a/logos_delivery/waku/waku_rendezvous/waku_peer_record.nim b/logos_delivery/waku/waku_rendezvous/waku_peer_record.nim index d6e700eb5..d0207ea00 100644 --- a/logos_delivery/waku/waku_rendezvous/waku_peer_record.nim +++ b/logos_delivery/waku/waku_rendezvous/waku_peer_record.nim @@ -1,14 +1,8 @@ import std/times, sugar -import - libp2p/[ - protocols/rendezvous, - signed_envelope, - multicodec, - multiaddress, - protobuf/minprotobuf, - peerid, - ] +import protobuf_serialization, protobuf_serialization/pkg/results +import libp2p/[protocols/rendezvous, signed_envelope, multicodec, multiaddress, peerid] +import ../common/protobuf type WakuPeerRecord* = object # Considering only mix as of now, but we can keep extending this to include all capabilities part of Waku ENR @@ -17,6 +11,12 @@ type WakuPeerRecord* = object addresses*: seq[MultiAddress] mixKey*: string +type WakuPeerRecordPB {.proto2.} = object + peerId {.fieldNumber: 1, ext, required.}: PeerId + seqNo {.fieldNumber: 2, pint, required.}: uint64 + addresses {.fieldNumber: 3, ext.}: seq[MultiAddress] + mixKey {.fieldNumber: 4, required.}: string + proc payloadDomain*(T: typedesc[WakuPeerRecord]): string = $multiCodec("libp2p-custom-peer-record") @@ -32,36 +32,36 @@ proc init*( ): T = WakuPeerRecord(peerId: peerId, seqNo: seqNo, addresses: addresses, mixKey: mixKey) +proc decodeWakuPeerRecord(buffer: seq[byte]): ProtobufResult[WakuPeerRecord] = + var pb: WakuPeerRecordPB + try: + pb = Protobuf.decode(buffer, WakuPeerRecordPB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + + if pb.addresses.len == 0: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + + ok( + WakuPeerRecord( + peerId: pb.peerId, seqNo: pb.seqNo, addresses: pb.addresses, mixKey: pb.mixKey + ) + ) + proc decode*( T: typedesc[WakuPeerRecord], buffer: seq[byte] -): Result[WakuPeerRecord, ProtoError] = - let pb = initProtoBuffer(buffer) - var record = WakuPeerRecord() - - ?pb.getRequiredField(1, record.peerId) - ?pb.getRequiredField(2, record.seqNo) - discard ?pb.getRepeatedField(3, record.addresses) - - if record.addresses.len == 0: - return err(ProtoError.RequiredFieldMissing) - - ?pb.getRequiredField(4, record.mixKey) - - return ok(record) +): ProtobufResult[WakuPeerRecord] = + decodeWakuPeerRecord(buffer) proc encode*(record: WakuPeerRecord): seq[byte] = - var pb = initProtoBuffer() - - pb.write(1, record.peerId) - pb.write(2, record.seqNo) - - for address in record.addresses: - pb.write(3, address) - - pb.write(4, record.mixKey) - - pb.finish() - return pb.buffer + Protobuf.encode( + WakuPeerRecordPB( + peerId: record.peerId, + seqNo: record.seqNo, + addresses: record.addresses, + mixKey: record.mixKey, + ) + ) proc checkWakuPeerRecord*( _: WakuPeerRecord, spr: seq[byte], peerId: PeerId diff --git a/logos_delivery/waku/waku_store/client.nim b/logos_delivery/waku/waku_store/client.nim index 82b7f4c43..8139c3910 100644 --- a/logos_delivery/waku/waku_store/client.nim +++ b/logos_delivery/waku/waku_store/client.nim @@ -43,7 +43,7 @@ proc sendStoreRequest( req.requestId = generateRequestId(self.rng) let writeRes = catch: - await connection.writeLP(req.encode().buffer) + await connection.writeLP(req.encode()) if writeRes.isErr(): return err(StoreError(kind: ErrorCode.BAD_REQUEST, cause: writeRes.error.msg)) @@ -65,7 +65,7 @@ proc sendStoreRequest( let topic = req.pubsubTopic.get() if not self.storeMsgMetricsPerShard.hasKey(topic): self.storeMsgMetricsPerShard[topic] = 0 - self.storeMsgMetricsPerShard[topic] += float64(req.encode().buffer.len) + self.storeMsgMetricsPerShard[topic] += float64(req.encode().len) waku_relay_fleet_store_msg_size_bytes.inc( self.storeMsgMetricsPerShard[topic], labelValues = [topic] diff --git a/logos_delivery/waku/waku_store/protocol.nim b/logos_delivery/waku/waku_store/protocol.nim index bf34bb615..cee02ce56 100644 --- a/logos_delivery/waku/waku_store/protocol.nim +++ b/logos_delivery/waku/waku_store/protocol.nim @@ -11,7 +11,6 @@ import bearssl/rand, libp2p/crypto/crypto, libp2p/protocols/protocol, - libp2p/protobuf/minprotobuf, libp2p/stream/connection, metrics import @@ -50,7 +49,7 @@ proc handleQueryRequest( res.statusCode = uint32(ErrorCode.BAD_REQUEST) res.statusDesc = "decoding rpc failed: " & $error - return (res.encode().buffer, "not_parsed_requestId") + return (res.encode(), "not_parsed_requestId") let requestId = req.requestId @@ -67,7 +66,7 @@ proc handleQueryRequest( res.statusCode = uint32(error.kind) res.statusDesc = $error - return (res.encode().buffer, "not_parsed_requestId") + return (res.encode(), "not_parsed_requestId") res.requestId = requestId res.statusCode = 200 @@ -76,7 +75,7 @@ proc handleQueryRequest( info "sending store query response", peerId = requestor, requestId = requestId, messages = res.messages.len - return (res.encode().buffer, requestId) + return (res.encode(), requestId) proc initProtocolHandler(self: WakuStore) = let rejectReposnseBuffer = StoreQueryResponse( @@ -86,7 +85,7 @@ proc initProtocolHandler(self: WakuStore) = requestId: "N/A", statusCode: uint32(ErrorCode.TOO_MANY_REQUESTS), statusDesc: $ErrorCode.TOO_MANY_REQUESTS, - ).encode().buffer + ).encode() proc handler(conn: Connection, proto: string) {.async: (raises: [CancelledError]).} = var successfulQuery = false ## only consider the correct queries in metrics diff --git a/logos_delivery/waku/waku_store/rpc_codec.nim b/logos_delivery/waku/waku_store/rpc_codec.nim index 2dbb890cd..33fe77e25 100644 --- a/logos_delivery/waku/waku_store/rpc_codec.nim +++ b/logos_delivery/waku/waku_store/rpc_codec.nim @@ -1,213 +1,175 @@ {.push raises: [].} -import results, stew/arrayops +import std/sequtils, results, stew/arrayops +import protobuf_serialization, protobuf_serialization/pkg/results import ../common/[protobuf, paging], ../waku_core, ./common const DefaultMaxRpcSize* = -1 +type + WakuMessageKeyValuePB {.proto2.} = object + messageHash {.fieldNumber: 1, required.}: seq[byte] + message {.fieldNumber: 2, ext.}: Opt[WakuMessage] + pubsubTopic {.fieldNumber: 3.}: Opt[string] + + StoreQueryRequestPB {.proto2.} = object + requestId {.fieldNumber: 1, required.}: string + includeData {.fieldNumber: 2.}: Opt[bool] + pubsubTopic {.fieldNumber: 10.}: Opt[string] + contentTopics {.fieldNumber: 11.}: seq[string] + startTime {.fieldNumber: 12, sint.}: Opt[int64] + endTime {.fieldNumber: 13, sint.}: Opt[int64] + messageHashes {.fieldNumber: 20.}: seq[seq[byte]] + paginationCursor {.fieldNumber: 51.}: Opt[seq[byte]] + paginationForward {.fieldNumber: 52, pint.}: Opt[uint32] + paginationLimit {.fieldNumber: 53, pint.}: Opt[uint64] + + StoreQueryResponsePB {.proto2.} = object + requestId {.fieldNumber: 1, required.}: string + statusCode {.fieldNumber: 10, pint, required.}: uint32 + statusDesc {.fieldNumber: 11, required.}: string + messages {.fieldNumber: 20.}: seq[WakuMessageKeyValuePB] + paginationCursor {.fieldNumber: 51.}: Opt[seq[byte]] + +proc toHash(s: seq[byte]): WakuMessageHash = + var h: WakuMessageHash + discard copyFrom[byte](h, s) + h + ### Request ### -proc encode*(req: StoreQueryRequest): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, req.requestId) - pb.write3(2, uint32(req.includeData)) - - pb.write3(10, req.pubsubTopic) - - for contentTopic in req.contentTopics: - pb.write3(11, contentTopic) - - pb.write3( - 12, - req.startTime.map( - proc(time: int64): zint64 = - zint64(time) - ), - ) - pb.write3( - 13, - req.endTime.map( - proc(time: int64): zint64 = - zint64(time) - ), +proc encode*(req: StoreQueryRequest): seq[byte] = + Protobuf.encode( + StoreQueryRequestPB( + requestId: req.requestId, + includeData: Opt.some(req.includeData), + pubsubTopic: req.pubsubTopic, + contentTopics: req.contentTopics, + startTime: req.startTime.map( + proc(t: Timestamp): int64 = + int64(t) + ), + endTime: req.endTime.map( + proc(t: Timestamp): int64 = + int64(t) + ), + messageHashes: req.messageHashes.mapIt(@it), + paginationCursor: req.paginationCursor.map( + proc(h: WakuMessageHash): seq[byte] = + @h + ), + paginationForward: Opt.some(uint32(ord(req.paginationForward))), + paginationLimit: req.paginationLimit, + ) ) - for hash in req.messagehashes: - pb.write3(20, hash) +proc decodeStoreQueryRequest(buffer: seq[byte]): ProtobufResult[StoreQueryRequest] = + var pb: StoreQueryRequestPB + try: + pb = Protobuf.decode(buffer, StoreQueryRequestPB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) - pb.write3(51, req.paginationCursor) - pb.write3(52, uint32(req.paginationForward)) - pb.write3(53, req.paginationLimit) - - pb.finish3() - - return pb + ok( + StoreQueryRequest( + requestId: pb.requestId, + includeData: pb.includeData.get(false), + pubsubTopic: pb.pubsubTopic, + contentTopics: pb.contentTopics, + startTime: pb.startTime.map( + proc(t: int64): Timestamp = + Timestamp(t) + ), + endTime: pb.endTime.map( + proc(t: int64): Timestamp = + Timestamp(t) + ), + messageHashes: pb.messageHashes.mapIt(toHash(it)), + paginationCursor: pb.paginationCursor.map( + proc(s: seq[byte]): WakuMessageHash = + toHash(s) + ), + paginationForward: PagingDirection(pb.paginationForward.get(1'u32)), + paginationLimit: pb.paginationLimit, + ) + ) proc decode*( T: type StoreQueryRequest, buffer: seq[byte] ): ProtobufResult[StoreQueryRequest] = - var req = StoreQueryRequest() - let pb = initProtoBuffer(buffer) - - if not ?pb.getField(1, req.requestId): - return err(ProtobufError.missingRequiredField("request_id")) - - var inclData: uint32 - if not ?pb.getField(2, inclData): - req.includeData = false - else: - req.includeData = inclData > 0 - - var pubsubTopic: string - if not ?pb.getField(10, pubsubTopic): - req.pubsubTopic = Opt.none(string) - else: - req.pubsubTopic = Opt.some(pubsubTopic) - - var topics: seq[string] - if not ?pb.getRepeatedField(11, topics): - req.contentTopics = @[] - else: - req.contentTopics = topics - - var start: zint64 - if not ?pb.getField(12, start): - req.startTime = Opt.none(Timestamp) - else: - req.startTime = Opt.some(Timestamp(int64(start))) - - var endTime: zint64 - if not ?pb.getField(13, endTime): - req.endTime = Opt.none(Timestamp) - else: - req.endTime = Opt.some(Timestamp(int64(endTime))) - - var buffer: seq[seq[byte]] - if not ?pb.getRepeatedField(20, buffer): - req.messageHashes = @[] - else: - req.messageHashes = newSeqOfCap[WakuMessageHash](buffer.len) - for buf in buffer: - var hash: WakuMessageHash - discard copyFrom[byte](hash, buf) - req.messageHashes.add(hash) - - var cursor: seq[byte] - if not ?pb.getField(51, cursor): - req.paginationCursor = Opt.none(WakuMessageHash) - else: - var hash: WakuMessageHash - discard copyFrom[byte](hash, cursor) - req.paginationCursor = Opt.some(hash) - - var paging: uint32 - if not ?pb.getField(52, paging): - req.paginationForward = PagingDirection.default() - else: - req.paginationForward = PagingDirection(paging) - - var limit: uint64 - if not ?pb.getField(53, limit): - req.paginationLimit = Opt.none(uint64) - else: - req.paginationLimit = Opt.some(limit) - - return ok(req) + decodeStoreQueryRequest(buffer) ### Response ### -proc encode*(keyValue: WakuMessageKeyValue): ProtoBuffer = - var pb = initProtoBuffer() +proc toPB(kv: WakuMessageKeyValue): WakuMessageKeyValuePB = + # message + pubsubTopic: both or neither + if kv.message.isSome() and kv.pubsubTopic.isSome(): + WakuMessageKeyValuePB( + messageHash: @(kv.messageHash), message: kv.message, pubsubTopic: kv.pubsubTopic + ) + else: + WakuMessageKeyValuePB(messageHash: @(kv.messageHash)) - pb.write3(1, keyValue.messageHash) +proc encode*(keyValue: WakuMessageKeyValue): seq[byte] = + Protobuf.encode(toPB(keyValue)) - if keyValue.message.isSome() and keyValue.pubsubTopic.isSome(): - pb.write3(2, keyValue.message.get().encode()) - pb.write3(3, keyValue.pubsubTopic.get()) +proc encode*(res: StoreQueryResponse): seq[byte] = + Protobuf.encode( + StoreQueryResponsePB( + requestId: res.requestId, + statusCode: res.statusCode, + statusDesc: res.statusDesc, + messages: res.messages.mapIt(toPB(it)), + paginationCursor: res.paginationCursor.map( + proc(h: WakuMessageHash): seq[byte] = + @h + ), + ) + ) - pb.finish3() +proc fromPB(pb: WakuMessageKeyValuePB): WakuMessageKeyValue = + # message + pubsubTopic: both or neither + if pb.message.isSome() and pb.pubsubTopic.isSome(): + WakuMessageKeyValue( + messageHash: toHash(pb.messageHash), + message: pb.message, + pubsubTopic: pb.pubsubTopic, + ) + else: + WakuMessageKeyValue(messageHash: toHash(pb.messageHash)) - return pb - -proc encode*(res: StoreQueryResponse): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, res.requestId) - - pb.write3(10, res.statusCode) - pb.write3(11, res.statusDesc) - - for msg in res.messages: - pb.write3(20, msg.encode()) - - pb.write3(51, res.paginationCursor) - - pb.finish3() - - return pb +proc decodeWakuMessageKeyValue(buffer: seq[byte]): ProtobufResult[WakuMessageKeyValue] = + try: + ok(fromPB(Protobuf.decode(buffer, WakuMessageKeyValuePB))) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) proc decode*( T: type WakuMessageKeyValue, buffer: seq[byte] ): ProtobufResult[WakuMessageKeyValue] = - var keyValue = WakuMessageKeyValue() - let pb = initProtoBuffer(buffer) + decodeWakuMessageKeyValue(buffer) - var buf: seq[byte] - if not ?pb.getField(1, buf): - return err(ProtobufError.missingRequiredField("message_hash")) - else: - var hash: WakuMessageHash - discard copyFrom[byte](hash, buf) - keyValue.messagehash = hash +proc decodeStoreQueryResponse(buffer: seq[byte]): ProtobufResult[StoreQueryResponse] = + var pb: StoreQueryResponsePB + try: + pb = Protobuf.decode(buffer, StoreQueryResponsePB) + except SerializationError: + return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) - var proto: ProtoBuffer - var topic: string - if ?pb.getField(2, proto) and ?pb.getField(3, topic): - keyValue.message = Opt.some(?WakuMessage.decode(proto.buffer)) - keyValue.pubsubTopic = Opt.some(topic) - else: - keyValue.message = Opt.none(WakuMessage) - keyValue.pubsubTopic = Opt.none(string) - - return ok(keyValue) + ok( + StoreQueryResponse( + requestId: pb.requestId, + statusCode: pb.statusCode, + statusDesc: pb.statusDesc, + messages: pb.messages.mapIt(fromPB(it)), + paginationCursor: pb.paginationCursor.map( + proc(s: seq[byte]): WakuMessageHash = + toHash(s) + ), + ) + ) proc decode*( T: type StoreQueryResponse, buffer: seq[byte] ): ProtobufResult[StoreQueryResponse] = - var res = StoreQueryResponse() - let pb = initProtoBuffer(buffer) - - if not ?pb.getField(1, res.requestId): - return err(ProtobufError.missingRequiredField("request_id")) - - var code: uint32 - if not ?pb.getField(10, code): - return err(ProtobufError.missingRequiredField("status_code")) - else: - res.statusCode = code - - var desc: string - if not ?pb.getField(11, desc): - return err(ProtobufError.missingRequiredField("status_desc")) - else: - res.statusDesc = desc - - var buffer: seq[seq[byte]] - if not ?pb.getRepeatedField(20, buffer): - res.messages = @[] - else: - res.messages = newSeqOfCap[WakuMessageKeyValue](buffer.len) - for buf in buffer: - let msg = ?WakuMessageKeyValue.decode(buf) - res.messages.add(msg) - - var cursor: seq[byte] - if not ?pb.getField(51, cursor): - res.paginationCursor = Opt.none(WakuMessageHash) - else: - var hash: WakuMessageHash - discard copyFrom[byte](hash, cursor) - res.paginationCursor = Opt.some(hash) - - return ok(res) + decodeStoreQueryResponse(buffer) diff --git a/logos_delivery/waku/waku_store_sync/codec.nim b/logos_delivery/waku/waku_store_sync/codec.nim index a00de8e0b..187e9973d 100644 --- a/logos_delivery/waku/waku_store_sync/codec.nim +++ b/logos_delivery/waku/waku_store_sync/codec.nim @@ -2,6 +2,7 @@ import std/sequtils, stew/[leb128, byteutils] +import protobuf_serialization import ../common/protobuf, ../waku_core/message, ../waku_core/time, ./common const @@ -9,13 +10,8 @@ const VarIntLen = 9 AvgCapacity = 1000 -proc encode*(value: WakuMessageAndTopic): ProtoBuffer = - var pb = initProtoBuffer() - - pb.write3(1, value.pubsub) - pb.write3(2, value.message.encode()) - - return pb +proc encode*(value: WakuMessageAndTopic): seq[byte] = + Protobuf.encode(value) proc deltaEncode*(itemSet: ItemSet): seq[byte] = # 1 byte for resolved bool and 32 bytes hash plus 9 bytes varint per elements @@ -335,17 +331,11 @@ proc deltaDecode*(T: type RangesData, buffer: seq[byte]): Result[T, string] = return ok(payload) +proc decodeWakuMessageAndTopic(buffer: seq[byte]): ProtobufResult[WakuMessageAndTopic] = + try: + ok(Protobuf.decode(buffer, WakuMessageAndTopic)) + except SerializationError: + err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure)) + proc decode*(T: type WakuMessageAndTopic, buffer: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buffer) - - var pubsub: string - if not ?pb.getField(1, pubsub): - return err(ProtobufError.missingRequiredField("pubsub")) - - var proto: ProtoBuffer - if not ?pb.getField(2, proto): - return err(ProtobufError.missingRequiredField("msg")) - - let message = ?WakuMessage.decode(proto.buffer) - - return ok(WakuMessageAndTopic(pubsub: pubsub, message: message)) + decodeWakuMessageAndTopic(buffer) diff --git a/logos_delivery/waku/waku_store_sync/common.nim b/logos_delivery/waku/waku_store_sync/common.nim index 1c718d726..b2b0063be 100644 --- a/logos_delivery/waku/waku_store_sync/common.nim +++ b/logos_delivery/waku/waku_store_sync/common.nim @@ -1,6 +1,6 @@ {.push raises: [].} -import chronos, stew/[byteutils] +import chronos, stew/[byteutils], protobuf_serialization import ../waku_core @@ -33,9 +33,9 @@ type fingerprints*: seq[Fingerprint] # Range type fingerprint stored here in order itemSets*: seq[ItemSet] # Range type itemset stored here in order - WakuMessageAndTopic* = object - pubsub*: PubSubTopic - message*: WakuMessage + WakuMessageAndTopic* {.proto2.} = object + pubsub* {.fieldNumber: 1, required.}: PubSubTopic + message* {.fieldNumber: 2, ext, required.}: WakuMessage const EmptyFingerprint*: Fingerprint = [ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, diff --git a/logos_delivery/waku/waku_store_sync/transfer.nim b/logos_delivery/waku/waku_store_sync/transfer.nim index 5d20afb18..32871cf46 100644 --- a/logos_delivery/waku/waku_store_sync/transfer.nim +++ b/logos_delivery/waku/waku_store_sync/transfer.nim @@ -51,7 +51,7 @@ type SyncTransfer* = ref object of LPProtocol proc sendMessage( conn: Connection, payload: WakuMessageAndTopic ): Future[Result[void, string]] {.async.} = - let rawPayload = payload.encode().buffer + let rawPayload = payload.encode() total_bytes_exchanged.inc(rawPayload.len, labelValues = [Transfer, Sending]) diff --git a/tests/all_tests_waku.nim b/tests/all_tests_waku.nim index 5498db307..a0a63e231 100644 --- a/tests/all_tests_waku.nim +++ b/tests/all_tests_waku.nim @@ -4,6 +4,7 @@ import ./test_waku # Waku core test suite import + ./waku_core/test_message_codec, ./waku_core/test_namespaced_topics, ./waku_core/test_time, ./waku_core/test_message_digest, diff --git a/tests/common/test_all.nim b/tests/common/test_all.nim index 1070c34e4..4277a519a 100644 --- a/tests/common/test_all.nim +++ b/tests/common/test_all.nim @@ -3,7 +3,6 @@ import ./test_base64_codec, ./test_enr_builder, - ./test_protobuf_validation, ./test_sqlite_migrations, ./test_parse_size, ./test_requestratelimiter, diff --git a/tests/common/test_protobuf_validation.nim b/tests/common/test_protobuf_validation.nim deleted file mode 100644 index ea81f7ba2..000000000 --- a/tests/common/test_protobuf_validation.nim +++ /dev/null @@ -1,96 +0,0 @@ -{.used.} - -import testutils/unittests -import logos_delivery/waku/common/protobuf - -## Fixtures - -const MaxTestRpcFieldLen = 5 - -type TestRpc = object - testField*: string - -proc init(T: type TestRpc, field: string): T = - T(testField: field) - -proc encode(rpc: TestRpc): ProtoBuffer = - var pb = initProtoBuffer() - pb.write3(1, rpc.testField) - pb.finish3() - pb - -proc encodeWithBadFieldId(rpc: TestRpc): ProtoBuffer = - var pb = initProtoBuffer() - pb.write3(666, rpc.testField) - pb.finish3() - pb - -proc decode(T: type TestRpc, buf: seq[byte]): ProtobufResult[T] = - let pb = initProtoBuffer(buf) - - var field: string - if not ?pb.getField(1, field): - return err(ProtobufError.missingRequiredField("test_field")) - if field.len > MaxTestRpcFieldLen: - return err(ProtobufError.invalidLengthField("test_field")) - - ok(TestRpc.init(field)) - -## Tests - -suite "Waku Common - libp2p minprotobuf wrapper": - test "serialize and deserialize - valid length field": - ## Given - let field = "12345" - - let rpc = TestRpc.init(field) - - ## When - let encodedRpc = rpc.encode() - let decodedRpcRes = TestRpc.decode(encodedRpc.buffer) - - ## Then - check: - decodedRpcRes.isOk() - - let decodedRpc = decodedRpcRes.tryGet() - check: - decodedRpc.testField == field - - test "serialize and deserialize - missing required field": - ## Given - let field = "12345" - - let rpc = TestRpc.init(field) - - ## When - let encodedRpc = rpc.encodeWithBadFieldId() - let decodedRpcRes = TestRpc.decode(encodedRpc.buffer) - - ## Then - check: - decodedRpcRes.isErr() - - let error = decodedRpcRes.tryError() - check: - error.kind == ProtobufErrorKind.MissingRequiredField - error.field == "test_field" - - test "serialize and deserialize - invalid length field": - ## Given - let field = "123456" # field.len = MaxTestRpcFieldLen + 1 - - let rpc = TestRpc.init(field) - - ## When - let encodedRpc = rpc.encode() - let decodedRpcRes = TestRpc.decode(encodedRpc.buffer) - - ## Then - check: - decodedRpcRes.isErr() - - let error = decodedRpcRes.tryError() - check: - error.kind == ProtobufErrorKind.InvalidLengthField - error.field == "test_field" diff --git a/tests/incentivization/test_rpc_codec.nim b/tests/incentivization/test_rpc_codec.nim index 0067f37ee..d01c7f8f1 100644 --- a/tests/incentivization/test_rpc_codec.nim +++ b/tests/incentivization/test_rpc_codec.nim @@ -10,13 +10,13 @@ suite "Waku Incentivization Eligibility Codec": let txHashAsBytes = @(txHash.bytes()) let eligibilityProof = EligibilityProof(proofOfPayment: Opt.some(txHashAsBytes)) let encoded = encode(eligibilityProof) - let decoded = EligibilityProof.decode(encoded.buffer).get() + let decoded = EligibilityProof.decode(encoded).get() check: eligibilityProof == decoded asyncTest "encode eligibility status": let eligibilityStatus = init(EligibilityStatus, true) let encoded = encode(eligibilityStatus) - let decoded = EligibilityStatus.decode(encoded.buffer).get() + let decoded = EligibilityStatus.decode(encoded).get() check: eligibilityStatus == decoded diff --git a/tests/node/peer_manager/peer_store/test_waku_peer_storage.nim b/tests/node/peer_manager/peer_store/test_waku_peer_storage.nim index 2f1a536c6..5c3dee66a 100644 --- a/tests/node/peer_manager/peer_store/test_waku_peer_storage.nim +++ b/tests/node/peer_manager/peer_store/test_waku_peer_storage.nim @@ -1,5 +1,5 @@ import - std/[nativesockets, sequtils], + std/[nativesockets, net, sequtils], testutils/unittests, libp2p/[multiaddress, peerid], libp2p/crypto/crypto, @@ -79,10 +79,7 @@ suite "Protobuf Serialisation": # Then the encoded RemotePeerInfo should be equal to the expected bytes check: - encodedRemotePeerInfo.buffer == expectedBuffer - encodedRemotePeerInfo.offset == 152 - encodedRemotePeerInfo.length == 0 - encodedRemotePeerInfo.maxSize == 4194304 + encodedRemotePeerInfo == expectedBuffer suite "decode": test "simple": diff --git a/tests/test_peer_manager.nim b/tests/test_peer_manager.nim index e2f5a5a26..c065b8676 100644 --- a/tests/test_peer_manager.nim +++ b/tests/test_peer_manager.nim @@ -11,7 +11,6 @@ import eth/p2p/discoveryv5/enr, eth/common/eth_types, libp2p/[builders, switch, multiaddress], - libp2p/protobuf/minprotobuf, libp2p/stream/[bufferstream, connection], libp2p/crypto/crypto, libp2p/protocols/pubsub/pubsub, diff --git a/tests/test_utils_compat.nim b/tests/test_utils_compat.nim index aa353f8e5..592f795ea 100644 --- a/tests/test_utils_compat.nim +++ b/tests/test_utils_compat.nim @@ -20,7 +20,7 @@ suite "Waku Payload": ## When let pb = msg.encode() - let msgDecoded = WakuMessage.decode(pb.buffer) + let msgDecoded = WakuMessage.decode(pb) ## Then check: @@ -41,7 +41,7 @@ suite "Waku Payload": ## When let pb = msg.encode() - let msgDecoded = WakuMessage.decode(pb.buffer) + let msgDecoded = WakuMessage.decode(pb) ## Then check: diff --git a/tests/test_waku_protobufs.nim b/tests/test_waku_protobufs.nim index 5902b46b4..919e6538e 100644 --- a/tests/test_waku_protobufs.nim +++ b/tests/test_waku_protobufs.nim @@ -11,22 +11,23 @@ procSuite "Waku Protobufs": # TODO: Missing test coverage in many encode/decode protobuf functions test "WakuMetadataResponse": - let res = WakuMetadataResponse(clusterId: Opt.some(7), shards: @[10, 23, 33]) + let res = + WakuMetadataResponse(clusterId: Opt.some(7'u32), shards: @[10'u32, 23, 33]) let buffer = res.encode() - let decodedBuff = WakuMetadataResponse.decode(buffer.buffer) + let decodedBuff = WakuMetadataResponse.decode(buffer) check: decodedBuff.isOk() decodedBuff.get().clusterId.get() == res.clusterId.get() decodedBuff.get().shards == res.shards test "WakuMetadataRequest": - let req = WakuMetadataRequest(clusterId: Opt.some(5), shards: @[100, 2, 0]) + let req = WakuMetadataRequest(clusterId: Opt.some(5'u32), shards: @[100'u32, 2, 0]) let buffer = req.encode() - let decodedBuff = WakuMetadataRequest.decode(buffer.buffer) + let decodedBuff = WakuMetadataRequest.decode(buffer) check: decodedBuff.isOk() decodedBuff.get().clusterId.get() == req.clusterId.get() diff --git a/tests/waku_archive/test_driver_queue_pagination.nim b/tests/waku_archive/test_driver_queue_pagination.nim index ddfe0e1c3..b95cb408f 100644 --- a/tests/waku_archive/test_driver_queue_pagination.nim +++ b/tests/waku_archive/test_driver_queue_pagination.nim @@ -1,7 +1,6 @@ {.used.} -import - results, std/[sequtils, algorithm], testutils/unittests, libp2p/protobuf/minprotobuf +import results, std/[sequtils, algorithm], testutils/unittests import logos_delivery/waku/[ waku_archive, diff --git a/tests/waku_core/test_all.nim b/tests/waku_core/test_all.nim index f7f4fad38..6eb12b606 100644 --- a/tests/waku_core/test_all.nim +++ b/tests/waku_core/test_all.nim @@ -1,6 +1,7 @@ {.used.} import + ./test_message_codec, ./test_message_digest, ./test_namespaced_topics, ./test_peers, diff --git a/tests/waku_core/test_message_codec.nim b/tests/waku_core/test_message_codec.nim new file mode 100644 index 000000000..d56fc2275 --- /dev/null +++ b/tests/waku_core/test_message_codec.nim @@ -0,0 +1,77 @@ +{.used.} + +import std/sequtils, testutils/unittests +import logos_delivery/waku/waku_core/message/[message, codec] +import logos_delivery/waku/common/protobuf + +const + Payload = @[byte 0x68, 0x69] # "hi" + ContentTopic = "/a/1/b/c" + +# all-default optionals +const OldDefaultsBytes = @[ + byte 0x0a, 0x02, 0x68, 0x69, 0x12, 0x08, 0x2f, 0x61, 0x2f, 0x31, 0x2f, 0x62, 0x2f, + 0x63, 0x18, 0x00, 0x50, 0x00, 0x5a, 0x00, 0xaa, 0x01, 0x00, 0xf8, 0x01, 0x00, +] + +# all fields set +const OldFullBytes = @[ + byte 0x0a, 0x02, 0x68, 0x69, 0x12, 0x08, 0x2f, 0x61, 0x2f, 0x31, 0x2f, 0x62, 0x2f, + 0x63, 0x18, 0x01, 0x50, 0xa4, 0x8b, 0xb0, 0x99, 0x09, 0x5a, 0x02, 0xaa, 0xbb, 0xaa, + 0x01, 0x01, 0xcc, 0xf8, 0x01, 0x01, +] + +# a proto3 peer drops defaulted fields; must still decode +const CanonicalDefaultsBytes = @[ + byte 0x0a, 0x02, 0x68, 0x69, 0x12, 0x08, 0x2f, 0x61, 0x2f, 0x31, 0x2f, 0x62, 0x2f, + 0x63, +] + +proc defaultsMsg(): WakuMessage = + WakuMessage(payload: Payload, contentTopic: ContentTopic) + +proc fullMsg(): WakuMessage = + WakuMessage( + payload: Payload, + contentTopic: ContentTopic, + version: 1, + timestamp: 1234567890, + meta: @[byte 0xaa, 0xbb], + proof: @[byte 0xcc], + ephemeral: true, + ) + +suite "Waku Message - codec wire format": + test "encode all-default optionals": + check defaultsMsg().encode() == OldDefaultsBytes + + test "encode all fields set": + check fullMsg().encode() == OldFullBytes + + test "decode all-default-optionals bytes": + let decoded = WakuMessage.decode(OldDefaultsBytes) + check decoded.isOk() + check decoded.get() == defaultsMsg() + + test "decode all-fields bytes": + let decoded = WakuMessage.decode(OldFullBytes) + check decoded.isOk() + check decoded.get() == fullMsg() + + test "decode tolerates omitted defaults": + let decoded = WakuMessage.decode(CanonicalDefaultsBytes) + check decoded.isOk() + check decoded.get() == defaultsMsg() + + test "encode/decode round-trips": + for msg in [defaultsMsg(), fullMsg()]: + let decoded = WakuMessage.decode(msg.encode()) + check decoded.isOk() + check decoded.get() == msg + + test "decode rejects meta field exceeding max length": + var msg = defaultsMsg() + msg.meta = toSeq(0.byte .. 66.byte) # 67 bytes > MaxMetaAttrLength (64) + let decoded = WakuMessage.decode(msg.encode()) + check decoded.isErr() + check decoded.error.kind == ProtobufErrorKind.InvalidLengthField diff --git a/tests/waku_core/test_message_digest.nim b/tests/waku_core/test_message_digest.nim index babf84f8f..8845c854a 100644 --- a/tests/waku_core/test_message_digest.nim +++ b/tests/waku_core/test_message_digest.nim @@ -143,8 +143,8 @@ suite "Waku Message - Deterministic hashing": meta = toSeq(0.byte .. 66.byte), # 67 bytes ) - let encodedInvalidMsg = message.encode - let decoded = WakuMessage.decode(encodedInvalidMsg.buffer) + let encodedInvalidMsg = message.encode() + let decoded = WakuMessage.decode(encodedInvalidMsg) check: decoded.isErr == true diff --git a/tests/waku_filter_v2/test_all.nim b/tests/waku_filter_v2/test_all.nim index 877795142..df56b864f 100644 --- a/tests/waku_filter_v2/test_all.nim +++ b/tests/waku_filter_v2/test_all.nim @@ -1,3 +1,3 @@ {.used.} -import ./test_waku_client, ./test_waku_filter_dos_protection +import ./test_waku_client, ./test_waku_filter_dos_protection, ./test_rpc_codec diff --git a/tests/waku_filter_v2/test_rpc_codec.nim b/tests/waku_filter_v2/test_rpc_codec.nim new file mode 100644 index 000000000..17046d62e --- /dev/null +++ b/tests/waku_filter_v2/test_rpc_codec.nim @@ -0,0 +1,46 @@ +{.used.} + +import results, testutils/unittests +import + logos_delivery/waku/waku_filter_v2/rpc, + logos_delivery/waku/waku_filter_v2/rpc_codec, + logos_delivery/waku/waku_core + +# ping request: requestId "x" (0a 01 78) + filterSubscribeType PING=0 (10 00) +const PingRequest = @[byte 0x0a, 0x01, 0x78, 0x10, 0x00] +# same ping with field 2 omitted, as a proto3 peer sends it +const PingRequestNoType = @[byte 0x0a, 0x01, 0x78] + +suite "Waku Filter v2 - codec wire format": + test "FilterSubscribeRequest emits the enum field even for ord 0 (PING)": + let req = FilterSubscribeRequest.ping("x") + check req.encode() == PingRequest + + test "decode of a request omitting field 2 defaults to SUBSCRIBER_PING": + let decoded = FilterSubscribeRequest.decode(PingRequestNoType) + check decoded.isOk() + check decoded.get().requestId == "x" + check decoded.get().filterSubscribeType == FilterSubscribeType.SUBSCRIBER_PING + + test "FilterSubscribeRequest round-trips (subscribe with topics)": + let req = FilterSubscribeRequest.subscribe( + "req-1", "/waku/2/rs/0/0", @["/a/1/b/c", "/d/2/e/f"] + ) + let decoded = FilterSubscribeRequest.decode(req.encode()) + check decoded.isOk() + check decoded.get() == req + + test "FilterSubscribeResponse round-trips": + let resp = FilterSubscribeResponse.ok("req-2") + let decoded = FilterSubscribeResponse.decode(resp.encode()) + check decoded.isOk() + check decoded.get() == resp + + test "MessagePush round-trips with nested WakuMessage": + let push = MessagePush( + wakuMessage: WakuMessage(payload: @[byte 1, 2, 3], contentTopic: "/a/1/b/c"), + pubsubTopic: "/waku/2/rs/0/0", + ) + let decoded = MessagePush.decode(push.encode()) + check decoded.isOk() + check decoded.get() == push diff --git a/tests/waku_lightpush/test_all.nim b/tests/waku_lightpush/test_all.nim index b5edd72fb..9c9a09b2f 100644 --- a/tests/waku_lightpush/test_all.nim +++ b/tests/waku_lightpush/test_all.nim @@ -1,3 +1,3 @@ {.used.} -import ./test_client, ./test_ratelimit +import ./test_client, ./test_ratelimit, ./test_rpc_codec diff --git a/tests/waku_lightpush/test_client.nim b/tests/waku_lightpush/test_client.nim index 3e8e1a359..5a2bbef6b 100644 --- a/tests/waku_lightpush/test_client.nim +++ b/tests/waku_lightpush/test_client.nim @@ -36,7 +36,7 @@ suite "Waku Lightpush Client": handler = proc( pubsubTopic: PubsubTopic, message: WakuMessage ): Future[WakuLightPushResult] {.async.} = - let msgLen = message.encode().buffer.len + let msgLen = message.encode().len if msgLen > int(DefaultMaxWakuMessageSize) + 64 * 1024: return lighpushErrorResult( LightPushErrorCode.PAYLOAD_TOO_LARGE, "length greater than maxMessageSize" diff --git a/tests/waku_lightpush/test_rpc_codec.nim b/tests/waku_lightpush/test_rpc_codec.nim new file mode 100644 index 000000000..92d098847 --- /dev/null +++ b/tests/waku_lightpush/test_rpc_codec.nim @@ -0,0 +1,46 @@ +{.used.} + +import results, testutils/unittests +import + logos_delivery/waku/waku_lightpush/rpc, + logos_delivery/waku/waku_lightpush/rpc_codec, + logos_delivery/waku/waku_core + +# response: requestId "x" (0a 01 78) + statusCode 200 in field 10 (50 c8 01) +const StatusResponse200 = @[byte 0x0a, 0x01, 0x78, 0x50, 0xc8, 0x01] + +suite "Waku Lightpush v3 - codec wire format": + test "LightPushResponse emits statusCode (distinct uint32)": + let resp = LightPushResponse( + requestId: "x", + statusCode: LightPushStatusCode(200), + statusDesc: Opt.none(string), + relayPeerCount: Opt.none(uint32), + ) + check resp.encode() == StatusResponse200 + + test "LightPushResponse round-trips with all fields": + let resp = LightPushResponse( + requestId: "req-1", + statusCode: LightPushStatusCode(429), + statusDesc: Opt.some("too many"), + relayPeerCount: Opt.some(3'u32), + ) + let decoded = LightPushResponse.decode(resp.encode()) + check decoded.isOk() + check decoded.get().requestId == resp.requestId + check decoded.get().statusCode == resp.statusCode + check decoded.get().statusDesc == resp.statusDesc + check decoded.get().relayPeerCount == resp.relayPeerCount + + test "LightpushRequest round-trips with nested WakuMessage": + let req = LightpushRequest( + requestId: "req-2", + pubSubTopic: Opt.some("/waku/2/rs/0/0"), + message: WakuMessage(payload: @[byte 1, 2, 3], contentTopic: "/a/1/b/c"), + ) + let decoded = LightpushRequest.decode(req.encode()) + check decoded.isOk() + check decoded.get().requestId == req.requestId + check decoded.get().pubSubTopic == req.pubSubTopic + check decoded.get().message == req.message diff --git a/tests/waku_lightpush_legacy/test_all.nim b/tests/waku_lightpush_legacy/test_all.nim index b5edd72fb..9c9a09b2f 100644 --- a/tests/waku_lightpush_legacy/test_all.nim +++ b/tests/waku_lightpush_legacy/test_all.nim @@ -1,3 +1,3 @@ {.used.} -import ./test_client, ./test_ratelimit +import ./test_client, ./test_ratelimit, ./test_rpc_codec diff --git a/tests/waku_lightpush_legacy/test_client.nim b/tests/waku_lightpush_legacy/test_client.nim index b6d406962..29dafeefc 100644 --- a/tests/waku_lightpush_legacy/test_client.nim +++ b/tests/waku_lightpush_legacy/test_client.nim @@ -37,7 +37,7 @@ suite "Waku Legacy Lightpush Client": handler = proc( pubsubTopic: PubsubTopic, message: WakuMessage ): Future[WakuLightPushResult[void]] {.async.} = - let msgLen = message.encode().buffer.len + let msgLen = message.encode().len if msgLen > int(DefaultMaxWakuMessageSize) + 64 * 1024: return err("length greater than maxMessageSize") handlerFuture.complete((pubsubTopic, message)) diff --git a/tests/waku_lightpush_legacy/test_rpc_codec.nim b/tests/waku_lightpush_legacy/test_rpc_codec.nim new file mode 100644 index 000000000..239651bd0 --- /dev/null +++ b/tests/waku_lightpush_legacy/test_rpc_codec.nim @@ -0,0 +1,57 @@ +{.used.} + +import results, testutils/unittests +import + logos_delivery/waku/waku_lightpush_legacy/rpc, + logos_delivery/waku/waku_lightpush_legacy/rpc_codec, + logos_delivery/waku/waku_core + +# PushResponse{isSuccess:false}: field 1 emitted even when false +const PushRespFalse = @[byte 0x08, 0x00] +# PushResponse{isSuccess:true}, info omitted +const PushRespTrue = @[byte 0x08, 0x01] +# PushResponse{isSuccess:false, info:"hi"} +const PushRespFalseInfo = @[byte 0x08, 0x00, 0x12, 0x02, 0x68, 0x69] +# PushRPC{requestId:"x"}, request/response omitted +const PushRpcIdOnly = @[byte 0x0a, 0x01, 0x78] + +suite "Waku Legacy Lightpush - codec wire format": + test "PushResponse always emits the required isSuccess field, even when false": + check PushResponse(isSuccess: false, info: Opt.none(string)).encode() == + PushRespFalse + + test "PushResponse success, no info": + check PushResponse(isSuccess: true, info: Opt.none(string)).encode() == PushRespTrue + + test "PushResponse failure, with info": + check PushResponse(isSuccess: false, info: Opt.some("hi")).encode() == + PushRespFalseInfo + + test "PushRPC omits absent optional request/response": + check PushRPC( + requestId: "x", request: Opt.none(PushRequest), response: Opt.none(PushResponse) + ).encode() == PushRpcIdOnly + + test "PushResponse round-trips": + for r in [ + PushResponse(isSuccess: false, info: Opt.none(string)), + PushResponse(isSuccess: true, info: Opt.some("done")), + ]: + let decoded = PushResponse.decode(r.encode()) + check decoded.isOk() + check decoded.get() == r + + test "PushRPC round-trips with nested request and message": + let rpc = PushRPC( + requestId: "req-1", + request: Opt.some( + PushRequest( + pubSubTopic: "/waku/2/rs/0/0", + message: WakuMessage(payload: @[byte 1, 2, 3], contentTopic: "/a/1/b/c"), + ) + ), + response: Opt.none(PushResponse), + ) + let decoded = PushRPC.decode(rpc.encode()) + check decoded.isOk() + check decoded.get() == rpc diff --git a/tests/waku_peer_exchange/test_protocol.nim b/tests/waku_peer_exchange/test_protocol.nim index fb77c3174..ed49b0b3e 100644 --- a/tests/waku_peer_exchange/test_protocol.nim +++ b/tests/waku_peer_exchange/test_protocol.nim @@ -330,7 +330,7 @@ suite "Waku Peer Exchange": let rpc = PeerExchangeRpc.makeRequest(1) var buffer: seq[byte] - await conn.writeLP(rpc.encode().buffer) + await conn.writeLP(rpc.encode()) buffer = await conn.readLp(DefaultMaxRpcSize.int) # Decode the response diff --git a/tests/waku_peer_exchange/test_rpc_codec.nim b/tests/waku_peer_exchange/test_rpc_codec.nim index fa61a8d2a..1181c7d36 100644 --- a/tests/waku_peer_exchange/test_rpc_codec.nim +++ b/tests/waku_peer_exchange/test_rpc_codec.nim @@ -23,7 +23,7 @@ suite "Peer Exchange RPC": asyncTest "Encode - Decode": # Setup let rpcReq = PeerExchangeRpc.makeRequest(2) - let rpcReqBuffer: seq[byte] = rpcReq.encode().buffer + let rpcReqBuffer: seq[byte] = rpcReq.encode() let resReq = PeerExchangeRpc.decode(rpcReqBuffer) check: @@ -47,7 +47,7 @@ suite "Peer Exchange RPC": let rpc = PeerExchangeRpc.makeResponse(peerInfos) # When encoding and decoding - let rpcBuffer: seq[byte] = rpc.encode().buffer + let rpcBuffer: seq[byte] = rpc.encode() let res = PeerExchangeRpc.decode(rpcBuffer) # Then the peerInfos match the originals diff --git a/tests/waku_relay/test_protocol.nim b/tests/waku_relay/test_protocol.nim index d3b94a996..da312c747 100644 --- a/tests/waku_relay/test_protocol.nim +++ b/tests/waku_relay/test_protocol.nim @@ -1053,7 +1053,7 @@ suite "Waku Relay": let msgWithoutPayload = fakeWakuMessage(contentTopic = contentTopic, payload = getByteSequence(0)) - sizeEmptyMsg = uint64(msgWithoutPayload.encode().buffer.len) + sizeEmptyMsg = uint64(msgWithoutPayload.encode().len) let msg1 = diff --git a/tests/waku_relay/utils.nim b/tests/waku_relay/utils.nim index 4d40ae8b1..04dea9c08 100644 --- a/tests/waku_relay/utils.nim +++ b/tests/waku_relay/utils.nim @@ -95,7 +95,7 @@ proc sendRlnMessageWithInvalidProof*( messageId = MessageId(0), ) let - rateLimitProof = rateLimitProofRes.get().encode().buffer + rateLimitProof = rateLimitProofRes.get().encode() message = WakuMessage(payload: @payload, contentTopic: contentTopic, proof: rateLimitProof) diff --git a/tests/waku_rln_relay/rln/waku_rln_relay_utils.nim b/tests/waku_rln_relay/rln/waku_rln_relay_utils.nim index 07c85922d..c2293f6c8 100644 --- a/tests/waku_rln_relay/rln/waku_rln_relay_utils.nim +++ b/tests/waku_rln_relay/rln/waku_rln_relay_utils.nim @@ -33,7 +33,7 @@ proc unsafeAppendRLNProof*( let proof = (waitFor manager.generateProof(msg.toRLNSignal(), epoch, messageId)).valueOr: return err("could not generate rln-v2 proof: " & $error) - msg.proof = proof.encode().buffer + msg.proof = proof.encode() return ok() proc getWakuRlnConfig*( diff --git a/tests/waku_rln_relay/test_all.nim b/tests/waku_rln_relay/test_all.nim index 706fff49e..1dca6c655 100644 --- a/tests/waku_rln_relay/test_all.nim +++ b/tests/waku_rln_relay/test_all.nim @@ -1,6 +1,7 @@ {.used.} import + ./test_rate_limit_proof_codec, ./test_rln_group_manager_onchain, ./test_waku_rln_relay, ./test_wakunode_rln_relay, diff --git a/tests/waku_rln_relay/test_rate_limit_proof_codec.nim b/tests/waku_rln_relay/test_rate_limit_proof_codec.nim new file mode 100644 index 000000000..d66d963c6 --- /dev/null +++ b/tests/waku_rln_relay/test_rate_limit_proof_codec.nim @@ -0,0 +1,39 @@ +{.used.} + +import testutils/unittests +import logos_delivery/waku/rln/protocol_types + +suite "RLN RateLimitProof codec": + test "encode/decode round-trips, field-preserving and byte-stable": + var proof: RateLimitProof + for i in 0 .. 127: + proof.proof[i] = byte(i) + for i in 0 .. 31: + proof.merkleRoot[i] = byte(i + 1) + proof.epoch[i] = byte(i + 2) + proof.shareX[i] = byte(i + 3) + proof.shareY[i] = byte(i + 4) + proof.nullifier[i] = byte(i + 5) + proof.rlnIdentifier[i] = byte(i + 6) + + let encoded = proof.encode() + let decoded = RateLimitProof.init(encoded) + + check decoded.isOk() + let d = decoded.get() + check: + d.proof == proof.proof + d.merkleRoot == proof.merkleRoot + d.epoch == proof.epoch + d.shareX == proof.shareX + d.shareY == proof.shareY + d.nullifier == proof.nullifier + d.rlnIdentifier == proof.rlnIdentifier + # re-encoding the decoded proof yields identical bytes + d.encode() == encoded + + test "all-zero proof round-trips": + let proof = RateLimitProof() + let decoded = RateLimitProof.init(proof.encode()) + check decoded.isOk() + check decoded.get().encode() == proof.encode() diff --git a/tests/waku_rln_relay/test_waku_rln_relay.nim b/tests/waku_rln_relay/test_waku_rln_relay.nim index d63c7c924..7a826f4de 100644 --- a/tests/waku_rln_relay/test_waku_rln_relay.nim +++ b/tests/waku_rln_relay/test_waku_rln_relay.nim @@ -124,7 +124,7 @@ suite "Waku rln relay": rlnIdentifier: rlnIdentifier, ) protobuf = rateLimitProof.encode() - decodednsp = RateLimitProof.init(protobuf.buffer) + decodednsp = RateLimitProof.init(protobuf) require: decodednsp.isOk() @@ -182,7 +182,7 @@ suite "Waku rln relay": let shareY3 = shareX3 proc encodeAndGetBuf(proof: RateLimitProof): seq[byte] = - return proof.encode().buffer + return proof.encode() let proof1 = RateLimitProof( diff --git a/tests/waku_rln_relay/utils_offchain.nim b/tests/waku_rln_relay/utils_offchain.nim index 2dbd31744..ad1369bac 100644 --- a/tests/waku_rln_relay/utils_offchain.nim +++ b/tests/waku_rln_relay/utils_offchain.nim @@ -77,7 +77,7 @@ proc sendRlnMessageWithInvalidProof*( client.rln.getCurrentEpoch(), ) let - rateLimitProof = rateLimitProofRes.get().encode().buffer + rateLimitProof = rateLimitProofRes.get().encode() message = WakuMessage(payload: @payload, contentTopic: contentTopic, proof: rateLimitProof) diff --git a/tests/waku_store/test_rpc_codec.nim b/tests/waku_store/test_rpc_codec.nim index 3821aea98..49bd5c8c0 100644 --- a/tests/waku_store/test_rpc_codec.nim +++ b/tests/waku_store/test_rpc_codec.nim @@ -24,7 +24,7 @@ procSuite "Waku Store - RPC codec": ## When let pb = query.encode() - let decodedQuery = StoreQueryRequest.decode(pb.buffer) + let decodedQuery = StoreQueryRequest.decode(pb) ## Then check: @@ -40,7 +40,7 @@ procSuite "Waku Store - RPC codec": ## When let pb = emptyQuery.encode() - let decodedEmptyQuery = StoreQueryRequest.decode(pb.buffer) + let decodedEmptyQuery = StoreQueryRequest.decode(pb) ## Then check: @@ -70,7 +70,7 @@ procSuite "Waku Store - RPC codec": ## When let pb = res.encode() - let decodedRes = StoreQueryResponse.decode(pb.buffer) + let decodedRes = StoreQueryResponse.decode(pb) ## Then check: @@ -86,7 +86,7 @@ procSuite "Waku Store - RPC codec": ## When let pb = emptyRes.encode() - let decodedEmptyRes = StoreQueryResponse.decode(pb.buffer) + let decodedEmptyRes = StoreQueryResponse.decode(pb) ## Then check: