feat(gossip): implement result reporting for payloads buffer push (#5443)

Signed-off-by: Shubham Singh <shubhsoch@gmail.com>

Author: FirePheonix

gossip/state/payloads_buffer.go

@@ -20,8 +20,8 @@ import (
 // sequence numbers. It also will provide the capability
 // to signal whenever expected block has arrived.
 type PayloadsBuffer interface {
-	// Adds new block into the buffer
-	Push(payload *proto.Payload)
+	// Adds new block into the buffer, returns true if added
+	Push(payload *proto.Payload) bool
 
 	// Returns next expected sequence number
 	Next() uint64
@@ -73,16 +73,16 @@ func (b *PayloadsBufferImpl) Ready() chan struct{} {
 // Push new payload into the buffer structure in case new arrived payload
 // sequence number is below the expected next block number payload will be
 // thrown away.
-// TODO return bool to indicate if payload was added or not, so that caller can log result.
-func (b *PayloadsBufferImpl) Push(payload *proto.Payload) {
+// Returns true if payload was added, false otherwise.
+func (b *PayloadsBufferImpl) Push(payload *proto.Payload) bool {
 	b.mutex.Lock()
 	defer b.mutex.Unlock()
 
 	seqNum := payload.SeqNum
 
 	if seqNum < b.next || b.buf[seqNum] != nil {
 		b.logger.Debugf("Payload with sequence number = %d has been already processed", payload.SeqNum)
-		return
+		return false
 	}
 
 	b.buf[seqNum] = payload
@@ -91,6 +91,7 @@ func (b *PayloadsBufferImpl) Push(payload *proto.Payload) {
 	if seqNum == b.next && len(b.readyChan) == 0 {
 		b.readyChan <- struct{}{}
 	}
+	return true
 }
 
 // Next function provides the number of the next expected block
@@ -153,9 +154,10 @@ type metricsBuffer struct {
 	chainID     string
 }
 
-func (mb *metricsBuffer) Push(payload *proto.Payload) {
-	mb.PayloadsBuffer.Push(payload)
+func (mb *metricsBuffer) Push(payload *proto.Payload) bool {
+	res := mb.PayloadsBuffer.Push(payload)
 	mb.reportSize()
+	return res
 }
 
 func (mb *metricsBuffer) Pop() *proto.Payload {

gossip/state/payloads_buffer_test.go

@@ -47,8 +47,8 @@ func TestPayloadsBufferImpl_Push(t *testing.T) {
 		t.Fatal("Wasn't able to generate random payload for test")
 	}
 
-	t.Log("Pushing new payload into buffer")
-	buffer.Push(payload)
+	t.Log("Pushing old payload into buffer (should return false)")
+	require.False(t, buffer.Push(payload))
 
 	// Payloads with sequence number less than buffer top
 	// index should not be accepted
@@ -64,8 +64,8 @@ func TestPayloadsBufferImpl_Push(t *testing.T) {
 		t.Fatal("Wasn't able to generate random payload for test")
 	}
 
-	t.Log("Pushing new payload into buffer")
-	buffer.Push(payload)
+	t.Log("Pushing valid payload into buffer (should return true)")
+	require.True(t, buffer.Push(payload))
 	t.Log("Getting next block sequence number")
 	require.Equal(t, buffer.Next(), uint64(5))
 	t.Log("Check block buffer size")
@@ -141,6 +141,9 @@ func TestPayloadsBufferImpl_ConcurrentPush(t *testing.T) {
 	require.Equal(t, int32(1), atomic.LoadInt32(&ready))
 	// Buffer size has to be only one
 	require.Equal(t, 1, buffer.Size())
+
+	// Check that we can't push it again
+	require.False(t, buffer.Push(payload))
 }
 
 // Tests the scenario where payload pushes and pops are interleaved after a Ready() signal.

gossip/state/state.go

@@ -775,7 +775,9 @@ func (s *GossipStateProviderImpl) addPayload(payload *proto.Payload, blockingMod
 		time.Sleep(enqueueRetryInterval)
 	}
 
-	s.payloads.Push(payload)
+	if !s.payloads.Push(payload) {
+		s.logger.Debugf("Payload with sequence number %d was not added to buffer (already processed or outdated)", payload.SeqNum)
+	}
 	s.logger.Debugf("Blocks payloads buffer size for channel [%s] is %d blocks", s.chainID, s.payloads.Size())
 	return nil
 }