fix(messaging): disambiguate sender-scoped packet IDs - #6624
Conversation
📝 WalkthroughWalkthroughChangesThe change replaces generated packet IDs in queued delivery with node-scoped persisted identities. It adds atomic claim and rollback operations, contact-scoped lookups, sender-aware duplicate handling, and conversation-aware reaction matching. Packet identity and delivery
Conversation-safe packet handling
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (5 passed)
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. Comment |
simulationstation
changed the title
jamesarich
left a comment
jamesarich
left a comment
There was a problem hiding this comment.
The core insight here is right — mesh packet IDs are not globally unique and treating them as such is a real defect class — and the WorkManager re-keying onto (myNodeNum, uuid) row identity is a solid improvement. I also confirmed this does not reintroduce the LazyColumn duplicate-key crash class: MessageListPaged keys on uuid (the Room rowid), reaction lists key on (user.id, emoji), and the packet_id index is already non-unique. Two things to fix before this can merge.
Blocking: the new belongsTo() filter hides inbound reactions on PKI direct messages
Packet.belongsTo() (core/database/.../entity/Packet.kt:92) gates on ContactKey(packet.contact_key).channel != channel, but those two values are derived through different paths:
Packet.contact_keyis built fromDataPacket.channel, whichMeshDataMapper.kt:44normalizes toNodeAddress.PKC_CHANNEL_INDEX(8) wheneverpki_encrypted == true.ReactionEntity.channelfor an inbound reaction is the rawMeshPacket.channel(MeshDataHandlerImpl.kt:604), which is 0 on a PKC DM —pki_encryptedis a separate flag, not the channel index.
So on any PKC direct message — the default for DMs on current firmware — the parent row has contact_key = "8!…" while the reaction row has channel = 0, belongsTo returns false, and the reaction is filtered out of toMessage().emojis. Because this filters at read time it applies to rows already in the database, so existing DM reactions disappear on upgrade rather than just new ones failing to land.
The asymmetry is visible in the code: locally-sent reactions store channel = ContactKey(contactKey).channel (= 8) at MessagingControllerImpl.kt:79, so your own reactions survive the filter and your peer's do not. The same root cause hits MeshDataHandlerImpl.kt:585 (val contactKey = "${packet.channel}$contactId"), so the parent lookup at :625 also misses and reaction notifications stop for DMs.
Suggested fix: derive the channel from dataMapper.toDataPacket(packet) in both places rather than re-deriving from the raw proto, so the PKC normalization is applied exactly once and consistently. Worth adding ReactionKeyTest cases at the PKC index — every current case uses a channel-0 contact key ("0^all", "0!aaaa0001"), which is why this passed.
Please replace the process-wide send mutex
SendMessageWorker.kt:96 adds a companion sendMutex that is acquired at :55 and held across radioController.sendMessage(packetData) at :67. Holding a mutex across a radio API call is a hazard this repo has been bitten by before (the firmware SWR wedge): if the send stalls waiting on queue status, every other SendMessageWorker blocks behind it until WorkManager's 10-minute window kills them. It's also asymmetric — DesktopMessageQueue has no equivalent serialization. Since the lock exists only to cover the legacy/UUID coexistence window, a compare-and-set of QUEUED → ENROUTE before the send (instead of after, at :68) gives the same no-double-send guarantee without a global lock.
Non-blocking, but please sign off explicitly
Several fail-closed choices have user-visible consequences worth stating in the description: an ambiguous handleAckNak (MeshDataHandlerImpl.kt:411) drops the ACK, leaving the message ENROUTE forever; reply parents scoped to contact_key means a reply quoting a message from another conversation loses its quote bubble; and a legacy WorkManager job with an ambiguous packet ID returns Result.failure(), so that queued message never sends and never retries.
CI
Your two red checks are not your fault — build-flatpak fails on a Gradle distribution checksum mismatch because main still pins 9.7.0 in scripts/verify-flatpak/desktop-offline.yaml (the #6611 rollback updated the wrapper but not the manifest). Your PR only trips it because it touches desktopApp/, which gates that job. #6625 fixes exactly those two lines and is in the merge queue now — rebase once it lands and both should clear.
137f80a to
e9b1306
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
desktopApp/src/main/kotlin/org/meshtastic/desktop/radio/DesktopMessageQueue.kt (1)
44-65: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winAtomically claim the row before sending.
Two
enqueuecalls can both read the same QUEUED row before either call updates its status. Both calls then send the packet. UseclaimQueuedPacketbeforesendMessage. Roll back only whensendMessagethrows.Proposed fix
- val packetData = - packetRepository.getPacketByPersistedId(persistedId) - ?: return@launch // Packet no longer exists in DB? Do not retry. + val claimed = packetRepository.claimQueuedPacket(persistedId) ?: return@launch + if (claimed.packet.status != MessageStatus.QUEUED) return@launch try { - radioController.sendMessage(packetData) - packetRepository.updateMessageStatus(persistedId, MessageStatus.ENROUTE) + radioController.sendMessage(claimed.packet) } catch (`@Suppress`("TooGenericExceptionCaught") e: Exception) { - Logger.w(e) { "Failed to send packet ${packetData.id}, re-queuing" } - packetRepository.updateMessageStatus(persistedId, MessageStatus.QUEUED) + Logger.w(e) { "Failed to send packet ${claimed.packet.id}, re-queuing" } + packetRepository.rollbackEnroutePacket(claimed.id) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@desktopApp/src/main/kotlin/org/meshtastic/desktop/radio/DesktopMessageQueue.kt` around lines 44 - 65, Update enqueue around claimQueuedPacket to atomically claim the queued packet before calling radioController.sendMessage, and return without sending when the claim fails because another enqueue already claimed it. Keep the existing success status update after sending, and update the status back to QUEUED only in the sendMessage exception path; do not roll back for connection checks, missing packets, or failed claims.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.kt`:
- Around line 613-616: Update the duplicate-reaction log in MeshDataHandlerImpl
to remove the fromId sender identifier and emoji payload from the message.
Retain only non-sensitive context such as packetId, replyId, and the existing
reaction count.
In
`@core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt`:
- Around line 123-125: Update the sendMessage flow to provide its errorEvents to
safeLaunch so the IllegalStateException from
PacketRepositoryImpl.insertRoomPacket is surfaced through the send-failure
snackbar. Preserve the existing input-clearing behavior and use the established
error event for failed sends.
In
`@core/data/src/commonTest/kotlin/org/meshtastic/core/data/repository/CommonPacketRepositoryTest.kt`:
- Around line 171-201: Make this test exercise a real concurrent race by
providing the repository with a multi-threaded IO dispatcher instead of the
single-threaded UnconfinedTestDispatcher, then invoke claimQueuedPacket and
claimQueuedPacketByPacketIdIfUnique through that racingRepository. Preserve the
existing assertions proving one caller performs the claim and the other observes
it, and ensure the test fails if the production claim operation is no longer
atomic.
In
`@core/service/src/androidMain/kotlin/org/meshtastic/core/service/AndroidMeshWorkerManager.kt`:
- Around line 41-46: In AndroidMeshWorkerManager’s enqueueUniqueWork call,
change the policy for the persisted send-message work name to
ExistingWorkPolicy.KEEP. Preserve the existing unique-name construction and
workRequest so repeated scheduling retains the active worker instead of
canceling and replacing it.
---
Outside diff comments:
In
`@desktopApp/src/main/kotlin/org/meshtastic/desktop/radio/DesktopMessageQueue.kt`:
- Around line 44-65: Update enqueue around claimQueuedPacket to atomically claim
the queued packet before calling radioController.sendMessage, and return without
sending when the claim fails because another enqueue already claimed it. Keep
the existing success status update after sending, and update the status back to
QUEUED only in the sendMessage exception path; do not roll back for connection
checks, missing packets, or failed claims.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: fe903cf2-0151-409a-a971-2748fa9dc0f2
📒 Files selected for processing (22)
core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.ktcore/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/PacketHandlerImpl.ktcore/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.ktcore/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImplTest.ktcore/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerTest.ktcore/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.ktcore/data/src/commonTest/kotlin/org/meshtastic/core/data/repository/CommonPacketRepositoryTest.ktcore/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.ktcore/database/src/commonMain/kotlin/org/meshtastic/core/database/entity/Packet.ktcore/database/src/commonTest/kotlin/org/meshtastic/core/database/entity/ReactionKeyTest.ktcore/repository/src/commonMain/kotlin/org/meshtastic/core/repository/MeshWorkerManager.ktcore/repository/src/commonMain/kotlin/org/meshtastic/core/repository/MessageQueue.ktcore/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketRepository.ktcore/repository/src/commonMain/kotlin/org/meshtastic/core/repository/usecase/SendMessageUseCase.ktcore/repository/src/commonTest/kotlin/org/meshtastic/core/repository/usecase/SendMessageUseCaseTest.ktcore/service/src/androidHostTest/kotlin/org/meshtastic/core/service/SendMessageWorkerTest.ktcore/service/src/androidMain/kotlin/org/meshtastic/core/service/AndroidMeshWorkerManager.ktcore/service/src/androidMain/kotlin/org/meshtastic/core/service/worker/SendMessageWorker.ktcore/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioController.ktdesktopApp/src/main/kotlin/org/meshtastic/desktop/radio/DesktopMessageQueue.ktdesktopApp/src/main/kotlin/org/meshtastic/desktop/stub/NoopStubs.ktfeature/messaging/src/androidMain/kotlin/org/meshtastic/feature/messaging/worker/WorkManagerMessageQueue.kt
|
Thanks — addressed both blockers in |
e9b1306 to
e45e191
Compare
|
The CodeRabbit follow-ups are addressed in @coderabbitai full review |
|
|
e45e191 to
c38f250
Compare
|
Rebased the branch as one logical commit ( I also re-audited the remaining CodeRabbit test-strength suggestions against the current tests. The requested end states are already pinned directly: the duplicate-reaction regression stubs the legacy singular lookup with an existing row and still requires insertion for the other sender; the reply regression inserts the wrong-conversation parent first and requires the correct parent in both flow and DAO lookup; and the PKI regressions require exclusion from both the opposite conversation and opposite channel while preserving legacy raw-channel rows. I did not add redundant assertions. Current-head validation passed the affected data/database/repository/service/messaging/Desktop test suites, @jamesarich, when convenient, would you please take another look? All requested code, limitation-documentation, test, and rebase items are now complete. |
c38f250 to
cab3ae9
Compare
|
Rebased again as one logical commit ( The combined implementation preserves sender/contact-scoped matching and atomic exact-row claim/rollback while carrying #6630's orphaned- Validation is green: affected database/data/repository/service/messaging/Desktop suites, current-head @jamesarich, the current head is ready for re-review when convenient. |
|
I hit what looks like the same class of bug, on a different internal queue, Setup: Heltec V3 (ESP32-S3), firmware 2.8.0, EU_868/LongFast, Wi-Fi OFF Symptom: messages sent from the phone get no delivery confirmation — From source review (firmware @ HEAD 2.8.0-dev): Same root mechanism as this issue (small fixed-size internal queue + |
|
Thanks — this is useful context, and it appears distinct from #6624. This PR fixes Android-side sender-scoped persistence and durable-send ownership; it cannot prevent a firmware The current firmware paths support the proposed mechanism: proxied MQTT downlinks are reinjected through the receive pipeline, and Your existing #6601 is the right Android-facing symptom tracker. A focused Update: the deterministic source-level regression and proposed priority invariant are now tracked in meshtastic/firmware#11439. |
cab3ae9 to
d94a2b6
Compare
|
Rebased the branch as one logical commit ( The only overlap was the foreground Validation is green: global Spotless/Detekt; affected data, repository, and messaging suites; exact database and Android worker regressions; Desktop tests/compile; KMP smoke compile; and @jamesarich, the current head is ready for re-review when convenient. |
❌ 1 Tests Failed:
View the top 1 failed test(s) by shortest run time
To view more test analytics, go to the Test Analytics Dashboard |
d94a2b6 to
acae64f
Compare
|
Rebased the existing single logical commit to Current-base validation is green with at most four workers: global Spotless and Detekt; @jamesarich, the current head is ready for re-review when convenient. |
jamesarich
left a comment
jamesarich
left a comment
There was a problem hiding this comment.
Re-reviewed against both blocking items. Both are resolved, and the reaction fix is better than what I proposed.
1. PKI reaction filter — fixed at the write path, plus a legacy read shim
I asked you to derive the channel through toDataPacket so the PKC normalization applies exactly once. That's what landed: contactKey = "${dataPacket.channel}$contactId" and the duplicate "${originalPacket.channel}$targetId" derivation are both gone, replaced by a single DataPacket.contactKey(myNodeNum) helper, and channel = packet.channel no longer feeds ReactionEntity. MeshDataHandlerTest now asserts NodeAddress.PKC_CHANNEL_INDEX is what actually gets persisted for an inbound PKI reaction, with the notification lookup resolving "8!remote" — so new rows agree with their parent's contact_key.
The part I hadn't asked for is the one that matters most: isLegacyInboundPkiChannel in belongsTo() keeps rows already in the database visible. My concern was that read-time filtering would make existing DM reactions vanish on upgrade, and fixing only the write path wouldn't have addressed that. The three ReactionKeyTest cases pin the shape exactly — normalized inbound PKI reaction attaches to its direct-message parent, pre-fix raw-channel inbound PKI reaction remains visible, and wrong non-PKI reaction channel remains excluded proving the shim didn't over-broaden into a general channel-mismatch bypass. Since the write path is fixed, the "legacy" name is accurate rather than aspirational.
2. Send mutex — replaced with an atomic row claim
The process-wide sendMutex is gone. claimQueuedPacket / claimQueuedPacketByPacketIdIfUnique do the QUEUED → ENROUTE transition in the DAO and return the pre-claim status, so the caller that observes QUEUED owns the send and a later caller observes ENROUTE and declines — the compare-and-set-before-send I suggested, but atomic in SQL rather than advisory in process memory, which is strictly better and works across the Desktop queue too.
a stalled radio send does not block an unrelated claimed row is the direct regression test for the wedge I was worried about, and queued row can be claimed for send only once, concurrent enqueue sends only the caller that owns the queued claim, send failure rolls back only the exact claimed row, and missing claim neither sends nor rolls back cover the rest of the state space.
3. Fail-closed sign-off
The "Explicit fail-closed limitations" section covers what I asked for, including the cross-conversation reply losing its quote bubble.
Before merge
Base is d8361ccd1; main is now b4bedd92f (#6670, #6640, #6618, #6663). GitHub reports no textual conflict, and the merge queue rebases and runs the merge_group gate on the merged result, so this is fine to enqueue — I just don't want the green tick on the current base read as verification of the merged state.
LGTM. Nicely done on both.
3 participants
Summary
ENROUTEtimeout and reconnect rearming while keying every timer and timeout update by the exact persisted rowRoot cause and impact
Mesh packet IDs are sender-scoped and are not suitable as globally unique database or work identifiers. Several paths previously relied on ID-only queries. In addition, the first revision compared a normalized PKI parent channel with a raw reaction channel, held a global mutex across the full radio send, replaced active persisted-row work, and left the Desktop sender on a non-atomic load/send/update sequence.
Those paths could discard a legitimate packet or reaction, hide PKI direct-message reactions, associate a reply/reaction with the wrong message, update the wrong delivery state, replace/load the wrong durable job, send the same durable Desktop row twice, or block unrelated workers behind a stalled radio call.
Behavior after this change
DataPacketcontact/channel, while legacy received PKI reactions stored on raw channel 0 remain visibleQUEUEDtoENROUTE; only the successful claimant sendsKEEP, preserving an already active requestENROUTE, so a concurrent ACK/NAK is not overwritten(myNodeNum, uuid), so rows sharing one mesh packet ID retain independent timers and only the exact still-ENROUTErow can time outSnackbarManager; normal input clearing remains unchangedExplicit fail-closed limitations
reply_id, not parent sender/conversation identity. A cross-conversation reply therefore omits the quote bubble under contact-scoped lookup, and same-conversation ID ambiguity remains unattached.packet_idmaps to zero or multiple rows fails without sending or retrying because no safe persisted row can be selected.Regression coverage
Added coverage for PKI channel-8 persistence and notification lookup, legacy raw-channel PKI rendering, opposite-conversation/channel exclusion, sender collisions, scoped/ambiguous parents, full-packet delivery matching, real multithreaded provider overlap around Room UUID/legacy claims, unique and ambiguous legacy jobs, exact-row failure rollback, ACK-race preservation, stalled unrelated sends, and two durable rows sharing a mesh packet ID. Additional regressions prove that WorkManager keeps the first active persisted-row request, the Desktop queue has one concurrent send owner and exact failure rollback, a missing Desktop claim does not send, send-persistence exceptions reach the messaging snackbar manager, an already-claimed row is selected unambiguously for timeout ownership, colliding mesh IDs receive independent rearmed timers, and timeout mutates only the targeted persisted row.
Validation
The rebased head passed global
spotlessCheckanddetekt; the affectedcore:data,core:repository, andfeature:messagingsuites; exactPacketDaoTest,ReactionKeyTest,SendMessageWorkerTest, andAndroidMeshWorkerManagerTestAndroid-host regressions; Desktop tests and compilation;kmpSmokeCompile; andgit diff --check, using at most four workers.Coordination
Rebased as one logical commit onto current
mainatd8361ccd, including #6664's Compose rollback.git range-difffrom the prior head is unchanged. The earlier rebase overlap in the foregroundmessage_sendanalytics path still retains upstream analytics after the exact persisted row is durably enqueued. This also keeps #6630's orphaned-ENROUTEtimeout/rearm behavior integrated with this PR's sender-scoped matching and atomic claim/rollback semantics. Draft PR #6598 remains complementary rather than a duplicate, but overlaps packet admission, queue lifecycle, and status persistence.Summary by CodeRabbit