Roll back dedup cache entry when inbound queue insertion fails

module/metrics.go

@@ -411,6 +411,8 @@ type NetworkCoreMetrics interface {
 	InboundMessageReceived(sizeBytes int, topic string, protocol string, messageType string)
 	// DuplicateInboundMessagesDropped increments the metric tracking the number of duplicate messages dropped by the node.
 	DuplicateInboundMessagesDropped(topic string, protocol string, messageType string)
+	// QueueFullInboundMessagesDropped increments the metric tracking the number of inbound messages dropped because the queue is full.
+	QueueFullInboundMessagesDropped(topic string, protocol string, messageType string)
 	// UnicastMessageSendingStarted increments the metric tracking the number of unicast messages sent by the node.
 	UnicastMessageSendingStarted(topic string)
 	// UnicastMessageSendingCompleted decrements the metric tracking the number of unicast messages sent by the node.

module/metrics/network.go

@@ -33,6 +33,7 @@ type NetworkCollector struct {
 	outboundMessageSize          *prometheus.HistogramVec
 	inboundMessageSize           *prometheus.HistogramVec
 	duplicateMessagesDropped     *prometheus.CounterVec
+	queueFullMessagesDropped     *prometheus.CounterVec
 	queueSize                    *prometheus.GaugeVec
 	queueDuration                *prometheus.HistogramVec
 	numMessagesProcessing        *prometheus.GaugeVec
@@ -111,6 +112,15 @@ func NewNetworkCollector(logger zerolog.Logger, opts ...NetworkCollectorOpt) *Ne
 		}, []string{LabelChannel, LabelProtocol, LabelMessage},
 	)
 
+	nc.queueFullMessagesDropped = promauto.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: namespaceNetwork,
+			Subsystem: subsystemGossip,
+			Name:      nc.prefix + "queue_full_messages_dropped",
+			Help:      "number of inbound messages dropped because the queue is full",
+		}, []string{LabelChannel, LabelProtocol, LabelMessage},
+	)
+
 	nc.dnsLookupDuration = promauto.NewHistogram(
 		prometheus.HistogramOpts{
 			Namespace: namespaceNetwork,
@@ -278,6 +288,12 @@ func (nc *NetworkCollector) DuplicateInboundMessagesDropped(topic, protocol, mes
 	nc.duplicateMessagesDropped.WithLabelValues(channels.NormalizeTopicForMetrics(topic), protocol, messageType).Add(1)
 }
 
+// QueueFullInboundMessagesDropped increments the metric tracking the number of inbound messages dropped because the queue is full.
+// Cluster topics are normalized to their prefix to prevent unbounded cardinality growth.
+func (nc *NetworkCollector) QueueFullInboundMessagesDropped(topic, protocol, messageType string) {
+	nc.queueFullMessagesDropped.WithLabelValues(channels.NormalizeTopicForMetrics(topic), protocol, messageType).Add(1)
+}
+
 func (nc *NetworkCollector) MessageAdded(priority int) {
 	nc.queueSize.WithLabelValues(strconv.Itoa(priority)).Inc()
 }

module/metrics/noop.go

@@ -28,6 +28,7 @@ func NewNoopCollector() *NoopCollector {
 func (nc *NoopCollector) OutboundMessageSent(int, string, string, string)        {}
 func (nc *NoopCollector) InboundMessageReceived(int, string, string, string)     {}
 func (nc *NoopCollector) DuplicateInboundMessagesDropped(string, string, string) {}
+func (nc *NoopCollector) QueueFullInboundMessagesDropped(string, string, string) {}
 func (nc *NoopCollector) UnicastMessageSendingStarted(topic string)              {}
 func (nc *NoopCollector) UnicastMessageSendingCompleted(topic string)            {}
 func (nc *NoopCollector) BlockProposed(*flow.Block)                              {}

network/cache/rcvcache.go

@@ -47,3 +47,8 @@ func NewReceiveCache(sizeLimit uint, opts ...func(cache *ReceiveCache)) *Receive
 func (r *ReceiveCache) Add(eventID []byte) bool {
 	return r.Backend.Add(flow.HashToID(eventID), struct{}{}) // ignore eviction status
 }
+
+// Remove removes the eventID from the cache. Returns true if the eventID was present and removed, false otherwise.
+func (r *ReceiveCache) Remove(eventID []byte) bool {
+	return r.Backend.Remove(flow.HashToID(eventID))
+}

network/cache/rcvcache_test.go

@@ -62,6 +62,21 @@ func (r *ReceiveCacheTestSuite) TestSingleElementAdd() {
 	assert.False(r.Suite.T(), r.c.Add(eventID3))
 }
 
+// TestRemove verifies that an event ID can be explicitly removed from the cache,
+// allowing future messages with the same event ID to be added again.
+func (r *ReceiveCacheTestSuite) TestRemove() {
+	eventID, err := message.EventId(channels.Channel("0"), []byte("event-1"))
+	require.NoError(r.T(), err)
+
+	assert.True(r.Suite.T(), r.c.Add(eventID))
+	assert.False(r.Suite.T(), r.c.Add(eventID))
+
+	assert.True(r.Suite.T(), r.c.Remove(eventID))
+	assert.False(r.Suite.T(), r.c.Remove(eventID))
+
+	assert.True(r.Suite.T(), r.c.Add(eventID))
+}
+
 // TestNoneExistence evaluates the correctness of cache operation against non-existing element
 func (r *ReceiveCacheTestSuite) TestNoneExistence() {
 	eventID, err := message.EventId(channels.Channel("1"), []byte("non-existing event"))

network/underlay/network.go

@@ -615,6 +615,18 @@ func (n *Network) processNetworkMessage(msg network.IncomingMessageScope) error
 	// insert the message in the queue
 	err := n.queue.Insert(qm)
 	if err != nil {
+		// Roll back the dedup cache entry so that a later retransmission of the same
+		// payload is not dropped as a duplicate because of a momentary queue-full
+		// window. The goroutine that successfully added the event ID exclusively owns
+		// the entry until this removal, so this cannot remove another goroutine's entry.
+		n.receiveCache.Remove(msg.EventID())
+
+		if errors.Is(err, queue.ErrQueueFull) {
+			// Queue-full drops are back-pressure events; record a metric so they are
+			// observable instead of only logged.
+			n.metrics.QueueFullInboundMessagesDropped(msg.Channel().String(), msg.Protocol().String(), msg.PayloadType())
+		}
+
 		return fmt.Errorf("failed to insert message in queue: %w", err)
 	}
 

network/underlay/network_test.go

@@ -1,18 +1,23 @@
 package underlay
 
 import (
+	"context"
 	"testing"
 
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/metrics"
 	modulemock "github.com/onflow/flow-go/module/mock"
 	"github.com/onflow/flow-go/network"
+	netcache "github.com/onflow/flow-go/network/cache"
+	"github.com/onflow/flow-go/network/channels"
 	"github.com/onflow/flow-go/network/message"
 	mockmsg "github.com/onflow/flow-go/network/mock"
 	p2plogging "github.com/onflow/flow-go/network/p2p/logging"
+	"github.com/onflow/flow-go/network/queue"
 	"github.com/onflow/flow-go/network/validator"
 	"github.com/onflow/flow-go/utils/unittest"
 )
@@ -145,3 +150,78 @@ func TestGetAuthorizedIdentity_ActivePeer(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, activeIdentity, identity)
 }
+
+// TestProcessNetworkMessage_QueueFullDropRollsBackDedupCache verifies that when a message is
+// dropped because the inbound queue is full, its dedup cache entry is rolled back, so a later
+// retransmission of the same payload is accepted. The dedup event ID is sender-independent
+// (a hash of channel and payload), so a stale entry would drop identical payloads from any sender.
+func TestProcessNetworkMessage_QueueFullDropRollsBackDedupCache(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel) // unblocks the queue's shutdown goroutine
+
+	metricsCollector := modulemock.NewNetworkCoreMetrics(t)
+
+	// Only the fields touched by processNetworkMessage (receiveCache, queue, metrics, logger)
+	// are needed; this mirrors a Network built with MessageQueueSize=1.
+	n := &Network{
+		logger:       zerolog.Nop(),
+		metrics:      metricsCollector,
+		receiveCache: netcache.NewReceiveCache(1000),
+		queue:        queue.NewMessageQueue(ctx, queue.GetEventPriority, metrics.NewNoopCollector(), 1),
+	}
+
+	channel := channels.ConsensusCommittee
+	payload := []byte("dropped-then-retransmitted-payload")
+
+	// Two scopes carrying the identical channel+payload but from different origins:
+	// originA's delivery attempt hits the queue-full window; originB is a different,
+	// honest publisher retransmitting the same payload afterwards.
+	newScope := func(origin flow.Identifier, p []byte) *message.IncomingMessageScope {
+		scope, err := message.NewIncomingScope(
+			origin,
+			message.ProtocolTypePubSub,
+			&message.Message{ChannelID: channel.String(), Payload: p},
+			p)
+		require.NoError(t, err)
+		return scope
+	}
+	victimFromA := newScope(unittest.IdentifierFixture(), payload)
+	victimFromB := newScope(unittest.IdentifierFixture(), payload)
+	filler := newScope(unittest.IdentifierFixture(), []byte("filler-payload"))
+
+	// The dedup event ID is sender-independent.
+	require.Equal(t, victimFromA.EventID(), victimFromB.EventID(),
+		"identical channel+payload from different senders yields the identical event ID")
+	require.NotEqual(t, filler.EventID(), victimFromA.EventID())
+
+	// Step 1: occupy the single queue slot.
+	require.NoError(t, n.processNetworkMessage(filler))
+	require.Equal(t, 1, n.queue.Len())
+
+	// Step 2: the victim message arrives during the queue-full window and is dropped.
+	metricsCollector.On("QueueFullInboundMessagesDropped", channel.String(), message.ProtocolTypePubSub.String(), victimFromA.PayloadType()).Once()
+	err := n.processNetworkMessage(victimFromA)
+	require.ErrorIs(t, err, queue.ErrQueueFull,
+		"victim message is rejected because the inbound queue is full")
+	require.Equal(t, 1, n.queue.Len(), "only the filler remains queued")
+
+	// Step 3: the failed enqueue rolled back the dedup cache entry, so the event ID of the
+	// dropped message is no longer present. Probing the cache with Add returns true (unseen).
+	require.True(t, n.receiveCache.Add(victimFromA.EventID()),
+		"event ID of a queue-full-dropped message must not remain in the dedup cache")
+
+	// The cache Add above is just a probe; remove it to restore the prior cache state.
+	require.True(t, n.receiveCache.Remove(victimFromA.EventID()),
+		"probe entry must be removable")
+
+	// Step 4: a queue worker drains the filler.
+	require.NotNil(t, n.queue.Remove())
+	require.Equal(t, 0, n.queue.Len())
+
+	// Step 5: the identical payload is retransmitted by a different publisher and accepted.
+	err = n.processNetworkMessage(victimFromB)
+	require.NoError(t, err,
+		"retransmission of a previously dropped message must be accepted")
+	require.Equal(t, 1, n.queue.Len(),
+		"retransmission must be enqueued so the engine can process it")
+}