Concurrency issues when reading and writing retain messages  #200

Activity

1. yh742

   changed the title [-]Concurrency Issues[/-] [+]Concurrency issues when reading and writing retain messages [/+] on Apr 17, 2023
2. mochi-co

   added

   needs investigationIdentify cause / reproduce issueIdentify cause / reproduce issue

   on Apr 21, 2023
3. mochi-co commented on Apr 21, 2023

   

   mochi-co

   on Apr 21, 2023

   Member

   More actions

   Hi @yh742 - could you paste the race condition logs so I can have a closer look? Thanks!
4. yh742 commented on Apr 22, 2023

   

   yh742

   on Apr 22, 2023

   Author

   More actions

   Sure, I have attached the log. For a bit of context, I'm trying to inject a retain message to a topic when a device connects and remove the retain message when disconnection occurs.

   WARNING: DATA RACE
   Write at 0x00c0000251e8 by goroutine 100:
     github.com/mochi-co/mqtt/v2.(*TopicsIndex).RetainMessage()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/topics.go:372 +0x37d
     github.com/mochi-co/mqtt/v2.(*Server).retainMessage()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:758 +0xe4
     github.com/mochi-co/mqtt/v2.(*Server).processPublish()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:715 +0x769
     github.com/mochi-co/mqtt/v2.(*Server).processPacket()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:569 +0x2a7
     github.com/mochi-co/mqtt/v2.(*Server).InjectPacket()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:658 +0xa4
     github.com/mochi-co/mqtt/v2.(*Server).Publish()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:641 +0x184
     github.com/mochi-co/mqtt/v2.(*Server).Publish-fm()
         <autogenerated>:1 +0x99
     github.com/yh742/halbroker/hook/auth/userpass.(*Hook).OnDisconnect()
         /home/sean/Documents/projects/i4/halbroker/hook/auth/userpass/hook.go:234 +0x673
     github.com/mochi-co/mqtt/v2.(*Hooks).OnDisconnect()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/hooks.go:237 +0x13b
     github.com/mochi-co/mqtt/v2.(*Server).attachClient()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:376 +0xc8c
     github.com/mochi-co/mqtt/v2.(*Server).EstablishConnection()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:312 +0x88
     github.com/mochi-co/mqtt/v2.(*Server).EstablishConnection-fm()
         <autogenerated>:1 +0x6d
     github.com/mochi-co/mqtt/v2/listeners.(*TCP).Serve.func1()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/listeners/tcp.go:84 +0x8c
   
   Previous read at 0x00c0000251e8 by goroutine 92:
     github.com/mochi-co/mqtt/v2.(*TopicsIndex).scanMessages()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/topics.go:455 +0x8da
     github.com/mochi-co/mqtt/v2.(*TopicsIndex).scanMessages()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/topics.go:463 +0xb72
     github.com/mochi-co/mqtt/v2.(*TopicsIndex).scanMessages()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/topics.go:463 +0xb72
     github.com/mochi-co/mqtt/v2.(*TopicsIndex).scanMessages()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/topics.go:463 +0xb72
     github.com/mochi-co/mqtt/v2.(*TopicsIndex).scanMessages()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/topics.go:463 +0xb72
     github.com/mochi-co/mqtt/v2.(*TopicsIndex).scanMessages()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/topics.go:471 +0x470
     github.com/mochi-co/mqtt/v2.(*TopicsIndex).Messages()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/topics.go:427 +0xdb
     github.com/mochi-co/mqtt/v2.(*Server).publishRetainedToClient()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:876 +0x90
     github.com/mochi-co/mqtt/v2.(*Server).processSubscribe()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:1055 +0x59c
     github.com/mochi-co/mqtt/v2.(*Server).processPacket()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:583 +0xf07
     github.com/mochi-co/mqtt/v2.(*Server).receivePacket()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:411 +0x87
     github.com/mochi-co/mqtt/v2.(*Server).receivePacket-fm()
         <autogenerated>:1 +0x89
     github.com/mochi-co/mqtt/v2.(*Client).Read()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/clients.go:349 +0x19c
     github.com/mochi-co/mqtt/v2.(*Server).attachClient()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:366 +0x9c4
     github.com/mochi-co/mqtt/v2.(*Server).EstablishConnection()
         /home/sean/go/pkg/mod/github.com/mochi-co/mqtt/v2@v2.2.7/server.go:312 +0x88
     github.com/mochi-co/mqtt/v2.(*Server).EstablishConnection-fm()
         <autogenerated>:1 +0x6d
     github.com/mochi-co/mqtt/v2/listeners.(*TCP).Serve.func1()
         /home/sean/go/pkg/m
   
5. mochi-co

   self-assigned this

   on May 4, 2023
6. thedevop commented on Jun 12, 2023

   

   thedevop

   on Jun 12, 2023

   Last edited by thedevop

   Collaborator

   More actions

   @mochi-co , although I haven't dive deep into the topics code, but after looking at it to add #236, I do see race conditions here.

   The issue is that although the particles and particle methods are protected by the lock, but subsequent operations on the result are not. Here are some examples:

   1. In TopicsIndex.scanMessages:
      https://github.com/mochi-co/mqtt/blob/af79b55b9f56ea49a09fea0450d3062ddbd0c13b/topics.go#L449-L464
      n.particles.getAll() is protected by lock, but the for loop itself is not. Some of the particle returned in the map maybe deleted while it is iterating through the map.
   2. Similar issue exist in TopicsIndex.scanSubscribers.

   Given both of these are only called by their respective exported method, adding lock in Messages and Subscribers as suggested by @yh742 can address this. There may be performance impact though.
7. mochi-co

   added

   discussionSomething to be discussedSomething to be discussed

   on Jun 19, 2023
8. mochi-co commented on Jun 19, 2023

   

   mochi-co

   on Jun 19, 2023

   Member

   More actions

   @thedevop I did suspect this may be the case when I wrote it but I was hoping that it would be rare enough not to cause an issue - apparently not.

   @yh742 I tried to replicate this but was unable to trigger the race condition error. It's possible I'm performing the operations in the wrong order or in insufficient scale. Do you (or @thedevop) have any example code (test or main.go) I can run to reproduce this? That would make it easier to determine the correct solution.

   Thanks!
9. mochi-co

   added

   help wantedExtra attention is neededExtra attention is needed

   on Sep 8, 2023
10. thedevop commented on Feb 25, 2025

    

    thedevop

    on Feb 25, 2025

    Collaborator

    More actions

    Close this until we can consistently reproduce the issue.
11. thedevop

    closed this as completedon Feb 25, 2025
12. ChristophNetsch

    added a commit that references this issue on Sep 2, 2026

    fix(topics): read particle.retainPath under the particle lock in scan…

    ...

    8b986f0