Roll back dedup cache entry when inbound queue insertion fails

[zhangchiqing]

Problem

processNetworkMessage adds an incoming message's event ID to the dedup (receive) cache before inserting it into the inbound queue. If the queue is full, insertion fails but the cache entry is never rolled back. Since the event ID is a hash of channel and payload only (no sender), a later retransmission of the same payload — from any publisher — is dropped as a duplicate and never reaches its engine. The entry eventually ages out via LRU eviction, so the effect is bounded, but the message is incorrectly treated as seen until then.

Queue-full drops are also currently only visible as a Warn log; there is no metric.

Changes

• ReceiveCache.Remove: new method so a dedup entry can be rolled back.
• processNetworkMessage: on queue insertion failure, remove the event ID from the dedup cache. The goroutine that successfully added the entry exclusively owns it until removal (concurrent duplicates get false from Add and drop), so the rollback cannot remove another goroutine's entry.
• New QueueFullInboundMessagesDropped metric (channel/protocol/message-type labels, cluster topics normalized) emitted on queue-full drops, making inbound back-pressure observable.

Tests

• network/cache: TestRemove covers the new ReceiveCache.Remove.
• network/underlay: regression test asserting that after a queue-full drop, the event ID does not remain in the dedup cache, the metric is emitted, and a retransmission of the same payload from a different publisher is accepted. Verified to fail at the rollback assertion without the fix; passes with -race.

---

^{Need help on this PR? Tag @codesmith-bot with what you need. Autofix is disabled.}

Summary by CodeRabbit

• New Features

  • Added metrics for inbound messages dropped when the receiving queue is full, categorized by channel, protocol, and message type.

• Bug Fixes

  • Messages rejected because the inbound queue is full can now be retransmitted successfully instead of being incorrectly treated as duplicates.
  • Queue-full drops are now recorded for improved monitoring and troubleshooting.

• Tests

  • Added coverage for queue-full handling, deduplication rollback, and retransmission behavior.

[github-actions]

Dependency Review

✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.

Scanned Files

None

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Advanced

Run ID: 5c76a49f-d5c6-4c0c-83db-6a7e15bea818

📥 Commits

Reviewing files that changed from the base of the PR and between 37e6f36 and d60a61e.

📒 Files selected for processing (9)

• module/metrics.go
• module/metrics/network.go
• module/metrics/noop.go
• module/mock/network_core_metrics.go
• module/mock/network_metrics.go
• network/cache/rcvcache.go
• network/cache/rcvcache_test.go
• network/underlay/network.go
• network/underlay/network_test.go

Included review availability: Your plan provides up to 4 included reviews per hour; 3 remain after this review.

---

📝 Walkthrough

Walkthrough

The change adds queue-full inbound drop metrics, implements the metric across collectors and mocks, and removes receive-cache entries when queue insertion fails so retransmissions can be accepted.

Changes

Queue-full inbound message handling

Layer / File(s)	Summary
Queue-full metric contract and implementations module/metrics.go, module/metrics/network.go, module/metrics/noop.go, module/mock/*	Adds QueueFullInboundMessagesDropped to the metrics interface, collectors, and generated mocks. The Prometheus collector records channel, protocol, and message-type labels.
Queue failure rollback and validation network/cache/rcvcache.go, network/cache/rcvcache_test.go, network/underlay/network.go, network/underlay/network_test.go	Adds receive-cache removal and calls it when queue insertion fails. Queue-full failures record the new metric. Tests verify cache rollback and retransmission acceptance.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Severity of issue fixed: Medium

Merge Risk: ⚪ Minimal · up to d60a6

The rollback and metric contracts are consistently implemented and covered by regression tests; no merge-blocking risk is established.

Sequence Diagram(s)

sequenceDiagram
  participant Network as processNetworkMessage
  participant ReceiveCache
  participant InboundQueue
  participant Metrics
  Network->>ReceiveCache: Add event ID
  Network->>InboundQueue: Insert inbound message
  InboundQueue-->>Network: queue.ErrQueueFull
  Network->>ReceiveCache: Remove event ID
  Network->>Metrics: QueueFullInboundMessagesDropped

Loading

Suggested reviewers: peterargue, turbolent

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 33.33% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff. Analyzed 3 functions across 9 files.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and concisely describes the primary change: rolling back the deduplication cache entry when inbound queue insertion fails.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

• Fix all pre-merge checks with AI

✨ Finishing Touches 💡 1

📝 Generate docstrings 💡

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch leo/dedup-cache-rollback

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[codecov-commenter]

Codecov Report

❌ Patch coverage is 11.94030% with 59 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
module/mock/network_core_metrics.go	0.00%	28 Missing ⚠️
module/mock/network_metrics.go	0.00%	28 Missing ⚠️
module/metrics/network.go	80.00%	2 Missing ⚠️
module/metrics/noop.go	0.00%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!