Set of just the msg hash is enough, fixes #156

eth/p2p/rlpx_protocols/waku_protocol.nim

@@ -104,7 +104,7 @@ type
     trusted*: bool
     wakuMode*: WakuMode
     topics*: seq[Topic]
-    received: HashSet[Message]
+    received: HashSet[Hash]
 
   P2PRequestHandler* = proc(peer: Peer, envelope: Envelope) {.gcsafe.}
 
@@ -272,9 +272,9 @@ p2pProtocol Waku(version = wakuVersion,
       # broadcasting this message. This too is seen here as a duplicate message
       # (see above comment). If we want to seperate these cases (e.g. when peer
       # rating), then we have to add a "peer.state.send" HashSet.
-      if peer.state.received.containsOrIncl(msg):
+      if peer.state.received.containsOrIncl(msg.hash):
         dropped_malicious_duplicate_envelopes.inc()
-        debug "Peer sending duplicate messages", peer, hash = msg.hash
+        trace "Peer sending duplicate messages", peer, hash = $msg.hash
         # await peer.disconnect(SubprotocolReason)
         continue
 
@@ -368,8 +368,8 @@ proc processQueue(peer: Peer) =
     wakuNet = peer.networkState(Waku)
 
   for message in wakuNet.queue.items:
-    if wakuPeer.received.contains(message):
-      # debug "message was already send to peer"
+    if wakuPeer.received.contains(message.hash):
+      # trace "message was already send to peer", hash = $message.hash, peer
       continue
 
     if message.pow < wakuPeer.powRequirement:
@@ -389,7 +389,7 @@ proc processQueue(peer: Peer) =
 
     trace "Adding envelope"
     envelopes.add(message.env)
-    wakuPeer.received.incl(message)
+    wakuPeer.received.incl(message.hash)
 
   if envelopes.len() > 0:
     trace "Sending envelopes", amount=envelopes.len
@@ -442,7 +442,7 @@ proc queueMessage(node: EthereumNode, msg: Message): bool =
   if not msg.allowed(wakuNet.config):
     return false
 
-  trace "Adding message to queue"
+  trace "Adding message to queue", hash = $msg.hash
   if wakuNet.queue[].add(msg):
     valid_envelopes.inc()
     # Also notify our own filters of the message we are sending,

eth/p2p/rlpx_protocols/whisper/whisper_types.nim

@@ -98,7 +98,7 @@ type
     ## room for those with higher pow, even if they haven't expired yet.
     ## Larger messages and those with high time-to-live will require more pow.
     items*: seq[Message] ## Sorted by proof-of-work
-    itemHashes*: HashSet[Message] ## For easy duplication checking
+    itemHashes*: HashSet[Hash] ## For easy duplication checking
     # XXX: itemHashes is added for easy message duplication checking and for
     # easy pruning of the peer received message sets. It does have an impact on
     # adding and pruning of items however.
@@ -486,7 +486,7 @@ proc initMessage*(env: Envelope, powCalc = true): Message =
     result.pow = calcPow(result.env.len.uint32, result.env.ttl, result.hash)
     trace "Message PoW", pow = result.pow.formatFloat(ffScientific)
 
-proc hash*(msg: Message): hashes.Hash = hash(msg.hash.data)
+proc hash*(hash: Hash): hashes.Hash = hashes.hash(hash.data)
 
 # NOTE: Hashing and leading zeroes calculation is now the same between geth,
 # parity and this implementation.
@@ -540,7 +540,7 @@ proc prune*(self: var Queue) {.raises: [].} =
       if pos != i:
         shallowCopy(self.items[pos], self.items[i])
       inc(pos)
-    else: self.itemHashes.excl(self.items[i])
+    else: self.itemHashes.excl(self.items[i].hash)
   setLen(self.items, pos)
 
 proc add*(self: var Queue, msg: Message): bool =
@@ -562,10 +562,10 @@ proc add*(self: var Queue, msg: Message): bool =
         return false
 
       self.items.del(self.items.len() - 1)
-      self.itemHashes.excl(last)
+      self.itemHashes.excl(last.hash)
 
   # check for duplicate
-  if self.itemHashes.containsOrIncl(msg):
+  if self.itemHashes.containsOrIncl(msg.hash):
     return false
   else:
     self.items.insert(msg, self.items.lowerBound(msg, cmpPow))

eth/p2p/rlpx_protocols/whisper_protocol.nim

@@ -80,7 +80,7 @@ type
     bloom*: Bloom
     isLightNode*: bool
     trusted*: bool
-    received: HashSet[Message]
+    received: HashSet[Hash]
 
   WhisperNetwork = ref object
     queue*: ref Queue
@@ -202,9 +202,9 @@ p2pProtocol Whisper(version = whisperVersion,
       # broadcasting this message. This too is seen here as a duplicate message
       # (see above comment). If we want to seperate these cases (e.g. when peer
       # rating), then we have to add a "peer.state.send" HashSet.
-      if peer.state.received.containsOrIncl(msg):
+      if peer.state.received.containsOrIncl(msg.hash):
         dropped_malicious_duplicate_envelopes.inc()
-        debug "Peer sending duplicate messages", peer, hash = msg.hash
+        trace "Peer sending duplicate messages", peer, hash = $msg.hash
         # await peer.disconnect(SubprotocolReason)
         continue
 
@@ -268,8 +268,8 @@ proc processQueue(peer: Peer) =
     whisperNet = peer.networkState(Whisper)
 
   for message in whisperNet.queue.items:
-    if whisperPeer.received.contains(message):
-      # debug "message was already send to peer"
+    if whisperPeer.received.contains(message.hash):
+      # trace "message was already send to peer", hash = $message.hash, peer
       continue
 
     if message.pow < whisperPeer.powRequirement:
@@ -283,7 +283,7 @@ proc processQueue(peer: Peer) =
 
     trace "Adding envelope"
     envelopes.add(message.env)
-    whisperPeer.received.incl(message)
+    whisperPeer.received.incl(message.hash)
 
   if envelopes.len() > 0:
     trace "Sending envelopes", amount=envelopes.len
@@ -335,7 +335,7 @@ proc queueMessage(node: EthereumNode, msg: Message): bool =
   if not msg.allowed(whisperNet.config):
     return false
 
-  trace "Adding message to queue"
+  trace "Adding message to queue", hash = $msg.hash
   if whisperNet.queue[].add(msg):
     valid_envelopes.inc()
     # Also notify our own filters of the message we are sending,