fixup: integration test failure

Author: CHr15F0x

nomos-da/network/core/src/protocols/replication/behaviour.rs

@@ -274,19 +274,24 @@ where
         peers
     }
 
+    fn replicate_message(&mut self, waker: &Waker, message: &DaMessage) {
+        let message_id = (message.blob.blob_id.to_vec(), message.subnetwork_id);
+        if self.seen_message_cache.contains(&message_id) {
+            return;
+        }
+        self.seen_message_cache.insert(message_id);
+        self.send_message_impl(Some(waker), message);
+    }
+
     /// Initiate sending a replication message **from outside the behaviour**
     pub fn send_message(&mut self, message: &DaMessage) {
+        let message_id = (message.blob.blob_id.to_vec(), message.subnetwork_id);
+        self.seen_message_cache.insert(message_id);
         let waker = self.waker.take();
         self.send_message_impl(waker.as_ref(), message);
     }
 
     fn send_message_impl(&mut self, waker: Option<&Waker>, message: &DaMessage) {
-        let message_id = (message.blob.blob_id.to_vec(), message.subnetwork_id);
-        if self.seen_message_cache.contains(&message_id) {
-            return;
-        }
-        self.seen_message_cache.insert(message_id);
-
         // Push a message in the queue for every single peer connected that is a member
         // of the selected subnetwork_id
         let peers = self.no_loopback_member_peers_of(message.subnetwork_id);
@@ -365,7 +370,7 @@ where
             Poll::Ready(Some(Ok((peer_id, message, read_half)))) => {
                 // Replicate the message to all connected peers from the same subnet if we
                 // haven't seen it yet
-                self.send_message_impl(Some(cx.waker()), &message);
+                self.replicate_message(cx.waker(), &message);
                 // Schedule waiting for any next incoming message on the same stream's read half
                 self.incoming_tasks
                     .push(Self::try_read_message(peer_id, read_half).boxed());