refactor: support dynamic MQTT subscriptions in `mqtt-channel`/`tedge-mqtt-ext` by jarhodes314 · Pull Request #3661 · thin-edge/thin-edge.io

jarhodes314 · 2025-06-02T13:25:46Z

Proposed changes

For the custom mapper, the user specifies which topics to subscribe the mapping pipelines to. In order to support dynamically reloading these pipelines, we need to be able to update the subscriptions. This was originally done in #3626, but this PR extracts out the changes for separate review.

To properly support dynamic subscriptions, the list of subscribers and their topic filters has been replaced with a trie. The general idea is that this structure maps topics to ids of subscribed clients, allowing us to efficiently retrieve the subscribed clients without checking every possible subscription. It manages which topics we subscribe to, ensuring we subscribe to a minimal set of topics to cover the currently subscribed clients, and (crucially) allows us to unsubscribe when topics are no longer consumed. Since this datastructure is separate, the goal is to have 100% unit test coverage here before merging (at the time of writing it is close to this, but not quite there).

How the trie is structured is detailed in

thin-edge.io/crates/extensions/tedge_mqtt_ext/src/trie.rs

Lines 7 to 77 in d8f3302

    
           /// A Trie for matching incoming MQTT messages with their subscribers 
        
           /// 
        
           /// # Structure 
        
           /// Each node of the trie contains: 
        
           /// - a list of subscribers to the current topic 
        
           /// - a map of segments to trie nodes 
        
           /// 
        
           /// As an example, if we subscribed a subscriber `"tedge-mapper"` to topic 
        
           /// `c8y/s/us`, the structure would look like: 
        
           /// 
        
           /// ```text 
        
           /// { 
        
           ///     "c8y": { 
        
           ///         subscribers: [], 
        
           ///         sub_nodes: { 
        
           ///             "s": { 
        
           ///                 subscribers: [], 
        
           ///                 sub_nodes: { 
        
           ///                     "us": { 
        
           ///                         subscribers: ["tedge-mapper"], 
        
           ///                         sub_nodes: {} 
        
           ///                     } 
        
           ///                 } 
        
           ///             } 
        
           ///         } 
        
           ///     } 
        
           /// } 
        
           /// ``` 
        
           /// 
        
           /// In practice, this is achieved by having a root node that always has no 
        
           /// subscribers. What is shown above is the `sub_nodes` field of the root node. 
        
           /// 
        
           /// # Subscription management 
        
           /// One of the requirements for the tedge MQTT actor is to manage subscriptions 
        
           /// from a bunch of clients and process them as a single MQTT channel client. 
        
           /// This means the actor maintains a minimal set of MQTT topic subscriptions to 
        
           /// cover all possible messages to its clients. 
        
           /// 
        
           /// For example, if `client-a` subscribes to `a/b/c` and `client-b` subscribes 
        
           /// to `a/#`, the actor should subscribe to `a/#` only, since this wildcard 
        
           /// topic captures all messages on `a/b/c`, so we don't require a separate 
        
           /// subscription. 
        
           /// 
        
           /// To allow the actor to subscribe/unsubscribe when appropriate, 
        
           /// [MqtTrie::insert] and [MqtTrie::remove] both return [SubscriptionDiff] 
        
           /// objects. This returns the subscribe/unsubscribe requests that need to be 
        
           /// made to the MQTT broker following the internal subscription change. 
        
           /// 
        
           /// Here are some examples of diffs that are returned 
        
           /// 
        
           /// ``` 
        
           /// # use tedge_mqtt_ext::trie::*; 
        
           /// 
        
           /// let mut t = MqtTrie::default(); 
        
           /// // First subscriber -> subscribe to that topic 
        
           /// assert_eq!(t.insert("a/b", 1), SubscriptionDiff { subscribe: ["a/b".into()].into(), unsubscribe: [].into() }); 
        
           /// // Another subscriber to the same topics -> don't need to change subscriptions 
        
           /// assert_eq!(t.insert("a/b", 2), SubscriptionDiff { subscribe: [].into(), unsubscribe: [].into() }); 
        
           /// // Subscriber to a different topic -> subscribe to that topic 
        
           /// assert_eq!(t.insert("a", 1), SubscriptionDiff { subscribe: ["a".into()].into(), unsubscribe: [].into() }); 
        
           /// // Subscriber to a segment wildcard -> subscribe to that topic, unsubscribe from static topic 
        
           /// assert_eq!(t.insert("a/+", 1), SubscriptionDiff { subscribe: ["a/+".into()].into(), unsubscribe: ["a/b".into()].into() }); 
        
           /// // Subscriber to a wildcard -> subscribe to that topic and unsubscribe from the matching ones 
        
           /// // Don't unsubscribe from the already unsubscribed a/b topic though 
        
           /// assert_eq!(t.insert("#", 1), SubscriptionDiff { subscribe: ["#".into()].into(), unsubscribe: ["a".into(), "a/+".into()].into() }); 
        
           /// ``` 
        
           /// 
        
           /// It is still possible to end up with overlapping subscriptions via this 
        
           /// method. For instance, `a/+/c` and `a/b/+` both subscribe to messages on 
        
           /// `a/b/c`, but aren't overlapping. Currently, [MqtTrie] handles this by 
        
           /// subscribing to both `a/+/c` and `a/b/+`.

One thing this PR doesn't do currently is handle retain messages. The effect of this is that if client is subscribed to a topic with a retain message and then another client subscribes, this new client does not receive the retain message since no MQTT subscription is created in this case.

Types of changes

Bugfix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Improvement (general improvements like code refactoring that doesn't explicitly fix a bug or add any new functionality)
Documentation Update (if none of the other choices apply)
Breaking change (fix or feature that would cause existing functionality to not work as expected)

Paste Link to the issue

Checklist

I have read the CONTRIBUTING doc
I have signed the CLA (in all commits with git commit -s. You can activate automatic signing by running just prepare-dev once)
I ran just format as mentioned in CODING_GUIDELINES
I used just check as mentioned in CODING_GUIDELINES
I have added tests that prove my fix is effective or that my feature works
I have added necessary documentation (if appropriate)

Further comments

codecov · 2025-06-02T13:35:39Z

Codecov Report

Attention: Patch coverage is 97.71739% with 42 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
crates/extensions/tedge_mqtt_ext/src/lib.rs	91.42%	23 Missing and 10 partials ⚠️
crates/common/mqtt_channel/src/connection.rs	90.38%	1 Missing and 4 partials ⚠️
crates/common/mqtt_channel/src/tests.rs	98.88%	1 Missing ⚠️
crates/common/mqtt_channel/src/topics.rs	83.33%	1 Missing ⚠️
crates/extensions/tedge_mqtt_ext/src/tests.rs	98.71%	0 Missing and 1 partial ⚠️
crates/extensions/tedge_mqtt_ext/src/trie.rs	99.91%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

github-actions · 2025-06-02T14:52:53Z

Robot Results

✅ Passed	❌ Failed	⏭️ Skipped	Total	Pass %	⏱️ Duration
651	0	3	651	100	1h57m42.192897s

crates/extensions/tedge_mqtt_ext/src/lib.rs

crates/common/mqtt_channel/src/lib.rs

didier-wenzek · 2025-06-02T15:41:23Z

crates/common/mqtt_channel/src/config.rs

    ///
    /// Default: An empty topic list
-    pub subscriptions: TopicFilter,
+    pub subscriptions: Arc<Mutex<TopicFilter>>,


As the subscriptions are never updated through the config but using the connection, I would consider to remove the use of interior mutability here, and to rather use these subscriptions as an initial value on connect.

I don't quite understand. The reason this is mutable is because we need to maintain the list of subscribed topics so we can resubscribe if the connection drops at any point.

crates/extensions/tedge_mqtt_ext/src/lib.rs

Bravo555 · 2025-06-03T07:35:27Z

tests/RobotFramework/tests/pkcs11/private_key_storage.robot


 *** Variables ***
-${CERT_TEMPLATE}     /etc/tedge/hsm/cert.template    # created by init_softhsm script
+${CERT_TEMPLATE}    /etc/tedge/hsm/cert.template    # created by init_softhsm script


I added commit resolving this to my PR here
2f09de6

Bravo555

As it is a standalone PR, I'd also like to see some tests, unless for some reason it's especially unfeasible to add them right now.

albinsuresh

Adding some integration tests in tests.rs to showcase the usage of the dynamic subscribe/unsubscribe feature would be good, as a system test probably isn't feasible in this state.

crates/extensions/tedge_mqtt_ext/src/lib.rs

didier-wenzek

The struct MqtTrie introduced here is the right abstraction to address dynamic MQTT subscriptions, although a bit involved. Nicely implemented with a good code coverage.

crates/extensions/tedge_mqtt_ext/src/trie.rs

didier-wenzek · 2025-06-06T09:20:50Z

crates/extensions/tedge_mqtt_ext/src/trie.rs

+        }
+    }
+
+    // TODO this should only accept valid message topic, not containing wildcards


In practice, this is check has already be done (because the message has been received from MQTT or wrapped into a Topic).

So I would simply enforced that at the top level: MqtTrie::matches<'a>(&'a self, topic: &Topic) -> Vec<&'a T> and adds a debug_assert! here on the head.

didier-wenzek · 2025-06-06T13:50:35Z

crates/extensions/tedge_mqtt_ext/src/trie.rs

+                (Some(_), None, _) => break None,
+                (None, None, Some(Winner::This)) => break Some(Ordering::Greater),
+                (None, None, Some(Winner::Other)) => break Some(Ordering::Less),
+                (None, None, None) => break Some(Ordering::Equal),


It's not easy to grasp the logic behind this long list of cases.

The tests help a lot, though.

But is my understand correct? Is the intent to define topic_filter_1 <= topic_filter_2 as "any topic accepted by topic_filter_1 is also accepted by topic_filter_2"?

I would add a doc comment to clarify this.

I added some missing tests 236a2ac

Still need to add a doc comment, although I have now grouped and commented the cases to broadly explain why we need them. Your understanding is correct though

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

This is an attempt to fix the existing issues with dynamic subscriptions, as well as hopefully improving the performance when matching messages against a large number of subscriptions. Still TODO: - Integrate this with the MQTT actor, so we can realise these benefits - Document the datastructure and its methods Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

…work, currently no way to actually enact the dynamic subscription changes from the client Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

…ribers Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

albinsuresh · 2025-06-19T21:18:24Z

crates/extensions/tedge_mqtt_ext/src/trie.rs

+        #[test]
+        fn issue_1() {
+            let mut t = MqtTrie::default();
+            t.insert("/#", 0);


Never realised that the root topic also could be empty, until seeing this test. What's weird is that both Topic::new("") and Topic::new("/") are valid topics, but neither Topic::new("/+") nor Topic::new("/#") are valid. And yet the trie accepts them. Hope that doesn't break any of the assumptions.

But that makes me wonder how the MQTT actor should react to invalid topics in the SubscriptionRequest, since SubscriptionDiff also stores the subscriptions as Hashset<String> and not Hashset<Topic>.

They aren't valid topics since they contain wildcards, but they are valid topic filters to subscribe to. So you can't publish a message to /# but you can subscribe to it and receive a message on e.g. /a or the empty topic name ``

but neither Topic::new("/+") nor Topic::new("/#") are valid

Aah yes, my bad. Wrongly used Topic instead of TopicFilter.

But that makes me wonder how the MQTT actor should react to invalid topics in the SubscriptionRequest

I see that the trie is still silently accepting bad topic filters like a/#/a but internally keeping it as a/#. We probably need to use stronger types like TopicFilter at least in the SubscriptionRequest to prevent that.

I see that the trie is silently accepting bad topic filters like a/#/a but internally keeping it as a/#. We probably need to use stronger types like TopicFilter to prevent that.

Yeah, but only at the entry point of the API i.e the 3 methods of MqtTrie.

See also for a similar comment: #3661 (comment)

This fixes the handling of subscriptions when we clone mqtt_channel::Config. With the previous behaviour, any `Config` clones shared all their subscriptions.

jarhodes314 requested review from albinsuresh, didier-wenzek and rina23q as code owners June 2, 2025 13:25

jarhodes314 had a problem deploying to Test Pull Request June 2, 2025 13:25 — with GitHub Actions Failure

jarhodes314 requested a review from a team as a code owner June 2, 2025 14:24

jarhodes314 temporarily deployed to Test Pull Request June 2, 2025 14:24 — with GitHub Actions Inactive

jarhodes314 temporarily deployed to Test Auto June 2, 2025 14:33 — with GitHub Actions Inactive

Bravo555 self-assigned this Jun 2, 2025

didier-wenzek reviewed Jun 2, 2025

View reviewed changes

Bravo555 reviewed Jun 2, 2025

View reviewed changes

crates/extensions/tedge_mqtt_ext/src/lib.rs Outdated Show resolved Hide resolved

Bravo555 reviewed Jun 3, 2025

View reviewed changes

Bravo555 removed their assignment Jun 3, 2025

albinsuresh reviewed Jun 3, 2025

View reviewed changes

crates/extensions/tedge_mqtt_ext/src/lib.rs Outdated Show resolved Hide resolved

crates/extensions/tedge_mqtt_ext/src/lib.rs Show resolved Hide resolved

crates/extensions/tedge_mqtt_ext/src/lib.rs Outdated Show resolved Hide resolved

reubenmiller added theme:mqtt Theme: mqtt and mosquitto related topics refactoring Developer value labels Jun 5, 2025

jarhodes314 had a problem deploying to Test Pull Request June 5, 2025 13:52 — with GitHub Actions Failure

jarhodes314 marked this pull request as draft June 5, 2025 13:53

jarhodes314 had a problem deploying to Test Pull Request June 5, 2025 14:51 — with GitHub Actions Failure

jarhodes314 had a problem deploying to Test Pull Request June 5, 2025 15:35 — with GitHub Actions Failure

jarhodes314 had a problem deploying to Test Pull Request June 5, 2025 16:00 — with GitHub Actions Failure

didier-wenzek had a problem deploying to Test Pull Request June 6, 2025 14:18 — with GitHub Actions Failure

didier-wenzek reviewed Jun 6, 2025

View reviewed changes

jarhodes314 temporarily deployed to Test Pull Request June 6, 2025 17:25 — with GitHub Actions Inactive

jarhodes314 had a problem deploying to Test Auto June 6, 2025 17:33 — with GitHub Actions Failure

jarhodes314 temporarily deployed to Test Pull Request June 6, 2025 17:40 — with GitHub Actions Inactive

jarhodes314 temporarily deployed to Test Auto June 6, 2025 17:49 — with GitHub Actions Inactive

jarhodes314 temporarily deployed to Test Pull Request June 9, 2025 10:07 — with GitHub Actions Inactive

jarhodes314 and others added 15 commits June 19, 2025 09:26

run formatter

b21d730

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Add some documentation to MqtTrie

6ba3c3e

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Fix handling of wildcard subscriptions and their parent topics

a026c4e

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Address missing coverage in trie implementation

df542b2

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

First pass at integrating trie with mqtt actor. Static subscriptions …

7a50289

…work, currently no way to actually enact the dynamic subscription changes from the client Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Add timeout to test to ensure it doesn't run indefinitely

ca5e089

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Add api to get client id for dynamic MQTT subscriptions

b4b0879

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Address remaining missing coverage

7d4aa2a

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Add unit tests for dynamic subscriptions in the actor

7817b18

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Improve naming, document filter ranking and fix bug in matching subsc…

034e04f

…ribers Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Remove rather than add the topic to the subscription list

e6866fe

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Fix tedge-agent subscriptions

0cc1598

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Add some more tests to cover some lines that weren't actually tested

b2c9f29

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Add property-based tests

2292145

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

jarhodes314 force-pushed the refactor/dynamic-mqtt-subs branch from 1c0ca94 to 2292145 Compare June 19, 2025 08:28

jarhodes314 temporarily deployed to Test Pull Request June 19, 2025 08:28 — with GitHub Actions Inactive

jarhodes314 had a problem deploying to Test Auto June 19, 2025 08:36 — with GitHub Actions Failure

jarhodes314 had a problem deploying to Test Auto June 19, 2025 08:59 — with GitHub Actions Failure

jarhodes314 temporarily deployed to Test Pull Request June 19, 2025 15:49 — with GitHub Actions Inactive

albinsuresh reviewed Jun 19, 2025

View reviewed changes

jarhodes314 had a problem deploying to Test Auto June 20, 2025 06:53 — with GitHub Actions Failure

Move subscription mutex from config to connection

1b249b3

This fixes the handling of subscriptions when we clone mqtt_channel::Config. With the previous behaviour, any `Config` clones shared all their subscriptions.

jarhodes314 force-pushed the refactor/dynamic-mqtt-subs branch from cf6005f to 1b249b3 Compare June 20, 2025 12:04

jarhodes314 temporarily deployed to Test Pull Request June 20, 2025 12:04 — with GitHub Actions Inactive

jarhodes314 temporarily deployed to Test Auto June 20, 2025 12:12 — with GitHub Actions Inactive

jarhodes314 added this pull request to the merge queue Jun 20, 2025

Merged via the queue into thin-edge:main with commit bcb8c1b Jun 20, 2025
34 checks passed

reubenmiller mentioned this pull request Jul 3, 2025

chore: fix clippy warnings #3720

Merged

11 tasks

	/// A Trie for matching incoming MQTT messages with their subscribers
	///
	/// # Structure
	/// Each node of the trie contains:
	/// - a list of subscribers to the current topic
	/// - a map of segments to trie nodes
	///
	/// As an example, if we subscribed a subscriber `"tedge-mapper"` to topic
	/// `c8y/s/us`, the structure would look like:
	///
	/// ```text
	/// {
	/// "c8y": {
	/// subscribers: [],
	/// sub_nodes: {
	/// "s": {
	/// subscribers: [],
	/// sub_nodes: {
	/// "us": {
	/// subscribers: ["tedge-mapper"],
	/// sub_nodes: {}
	/// }
	/// }
	/// }
	/// }
	/// }
	/// }
	/// ```
	///
	/// In practice, this is achieved by having a root node that always has no
	/// subscribers. What is shown above is the `sub_nodes` field of the root node.
	///
	/// # Subscription management
	/// One of the requirements for the tedge MQTT actor is to manage subscriptions
	/// from a bunch of clients and process them as a single MQTT channel client.
	/// This means the actor maintains a minimal set of MQTT topic subscriptions to
	/// cover all possible messages to its clients.
	///
	/// For example, if `client-a` subscribes to `a/b/c` and `client-b` subscribes
	/// to `a/#`, the actor should subscribe to `a/#` only, since this wildcard
	/// topic captures all messages on `a/b/c`, so we don't require a separate
	/// subscription.
	///
	/// To allow the actor to subscribe/unsubscribe when appropriate,
	/// [MqtTrie::insert] and [MqtTrie::remove] both return [SubscriptionDiff]
	/// objects. This returns the subscribe/unsubscribe requests that need to be
	/// made to the MQTT broker following the internal subscription change.
	///
	/// Here are some examples of diffs that are returned
	///
	/// ```
	/// # use tedge_mqtt_ext::trie::*;
	///
	/// let mut t = MqtTrie::default();
	/// // First subscriber -> subscribe to that topic
	/// assert_eq!(t.insert("a/b", 1), SubscriptionDiff { subscribe: ["a/b".into()].into(), unsubscribe: [].into() });
	/// // Another subscriber to the same topics -> don't need to change subscriptions
	/// assert_eq!(t.insert("a/b", 2), SubscriptionDiff { subscribe: [].into(), unsubscribe: [].into() });
	/// // Subscriber to a different topic -> subscribe to that topic
	/// assert_eq!(t.insert("a", 1), SubscriptionDiff { subscribe: ["a".into()].into(), unsubscribe: [].into() });
	/// // Subscriber to a segment wildcard -> subscribe to that topic, unsubscribe from static topic
	/// assert_eq!(t.insert("a/+", 1), SubscriptionDiff { subscribe: ["a/+".into()].into(), unsubscribe: ["a/b".into()].into() });
	/// // Subscriber to a wildcard -> subscribe to that topic and unsubscribe from the matching ones
	/// // Don't unsubscribe from the already unsubscribed a/b topic though
	/// assert_eq!(t.insert("#", 1), SubscriptionDiff { subscribe: ["#".into()].into(), unsubscribe: ["a".into(), "a/+".into()].into() });
	/// ```
	///
	/// It is still possible to end up with overlapping subscriptions via this
	/// method. For instance, `a/+/c` and `a/b/+` both subscribe to messages on
	/// `a/b/c`, but aren't overlapping. Currently, [MqtTrie] handles this by
	/// subscribing to both `a/+/c` and `a/b/+`.

Conversation

jarhodes314 commented Jun 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Proposed changes

Types of changes

Paste Link to the issue

Checklist

Further comments

Uh oh!

codecov bot commented Jun 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

github-actions bot commented Jun 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Robot Results

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Bravo555 left a comment

Choose a reason for hiding this comment

Uh oh!

albinsuresh left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

didier-wenzek left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jarhodes314 Jun 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

albinsuresh Jun 20, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

jarhodes314 commented Jun 2, 2025 •

edited

Loading

codecov bot commented Jun 2, 2025 •

edited

Loading

github-actions bot commented Jun 2, 2025 •

edited

Loading

jarhodes314 Jun 6, 2025 •

edited

Loading

albinsuresh Jun 20, 2025 •

edited

Loading