Fix race condition in MQTT protocol when sending messages

[yanmxa]

Signed-off-by: myan [email protected]

Resolved: #1093

Summary:

Reason: This issue occurs when a single client sends events across multiple goroutines. Specifically, in the code here:

sdk-go/protocol/mqtt_paho/v2/protocol.go

Lines 92 to 101 in 1aecb20

	msg := p.publishOption
	if cecontext.TopicFrom(ctx) != "" {
	msg.Topic = cecontext.TopicFrom(ctx)
	cecontext.WithTopic(ctx, "")
	}
	err = WritePubMessage(ctx, m, msg, transformers...)
	if err != nil {
	return err
	}

at line 98, the MQTT msg is designed to hold a single binding.Message (or Event) payload. However, in a multi-goroutine environment, multiple m values may be written to the shared msg, causing a panic like bytes.Buffer: too large.

Solution: Instead of sharing the same msg across goroutines, we will create a copy of it each time a message is sent.

[yanmxa]

/cc @embano1

[Cali0707]

If I'm understanding correctly, the race condition is that the publishOption can be changed before the message is published?

Since there is only one place to set this field in the struct (

sdk-go/protocol/mqtt_paho/v2/option.go

Lines 29 to 37 in e6a74ef

	func WithPublish(publishOpt *paho.Publish) Option {
	return func(p *Protocol) error {
	if publishOpt == nil {
	return fmt.Errorf("the paho.Publish option must not be nil")
	}
	p.publishOption = publishOpt
	return nil
	}
	}

), maybe it would make sense to make the copy there instead of every time a message is published?

[embano1]

+1

Left a comment in the issue to better understand the issue/root cause. What is causing the race? Why is publishOption changing during concurrent calls? Asking all these questions because I didn't write the implementation and not familiar with MQ. Furthermore, if we change as per the above, I wonder what the implication is for existing users because we're changing the semantics here (copy instead of pointer).

[yanmxa]

Thanks @Cali0707! The code block you referred to is only invoked once when initializing the client. The issues arose when sending messages in many goroutines. Here is the detail

[yanmxa]

FYI @embano1.

I append the root cause here.

The publishOption is a pointer reference by the MQTT message, which is used to hold the CloudEvent payload. The message(publishOption) is shared across goroutines and will be changed by them.

The publishOption is a pointer referenced by the MQTT message, which holds the CloudEvent payload. So the message (publishOption) is shared across multiple goroutines and can be modified by them.

By making the suggested changes, the user should not perceive anything.

[embano1]

and will be changed by them

Can you please point me to the code/line where this happens?
Also, ideally you also write a test please to surface the current issue (panics) and how this PR fixes it, perhaps there's more issues due to the current implementation which we can detect this way 🙏

[yanmxa]

@embano1 No worries!

Here’s the code line where the panic occurs.

I’ve also written an integration test that covers the concurrent panic issue, which you can use to reproduce it.

[embano1]

+1

Left a comment in the issue to better understand the issue/root cause. What is causing the race? Why is publishOption changing during concurrent calls? Asking all these questions because I didn't write the implementation and not familiar with MQ. Furthermore, if we change as per the above, I wonder what the implication is for existing users because we're changing the semantics here (copy instead of pointer).

[embano1]

Great, nice work! Please squash your commits and force-push.

[yanmxa]

Great, nice work! Please squash your commits and force-push.

Done.

[embano1]

Thank you!