Contributing new features, PersistentSessions/WillMessage/MessageExpiry/SessionExpiry handling

[logicaloud]

Is this a good time to contribute these features? There seem to be more issues relating to these features popping up, plus some that have been on the issue list for a while.

My preference would be to submit this as two or maybe three separate pull requests. Ideally the pull requests would not be hanging around for "too long" to avoid many ongoing merges since that could be time consuming. So, a good time would be when someone actually has time to review the changes and when it does not clash with other developments.

Any advice on how to best approach this?

[chkr1011]

I am thinking about this topic for a long time. Here are my thoughts:

• We should implement several features in one chunk (custom store for retained messages and expiry interval)
• We should remove the existing events and create a new Interface which allows replacing the actual "RetainedMessagesStore"
  • The store should have no method for loading the retained messages. This should be the responsibility of the implementation.
  • The store must also support the query for matching messages based on a topic on its own. We may need to extract the topic matching code in a static or whatever class so that other "providers" can make use of it. The reason for this is that a store like SQL database may want to load messages only which are matching the topic. With the current code the storage must always return all messages so that another source makes no sense in terms of performance optimization.
  • When we have a new store the question is who is responsible for removing expired messages? I thought just removing them when accessing the store but this would delete the messages only when there is traffic. Having a timer (or similar) which will remove the messages every second may produce high CPU load.
• The current strategy "in memory" store should be set as the default implementation (not in options) and the server should fail to start when there is no store implementation set.
• The APIs of the store should be more precise like "AddRetainedMessage(...)" or "RemoveRetainedMessage(...)" as you mentioned earlier.
• All methods of the store must be returning a Task and accept a Cancellation Token.

What is your opinion on that? I think the biggest effort will be extracting the code for a query into the store so that there is no longer the need to hold all messages in memory. But I am afraid that this will also have the most pitfalls regarding to performance issues.

[logicaloud]

Thank you for comments.

We should implement several features in one chunk (custom store for retained messages and expiry interval)

Yes, these features become somewhat entangled, i.e., message expiry, will message delay, session expiry, can all rely on the same logic that "expires them".

We should remove the existing events and create a new Interface which allows replacing the actual "RetainedMessagesStore"

The events work well when implementing a backing store. That means, at runtime all message are kept in memory like they are now, but whenever messages are added/removed/updated the existing events can inform the backing store that persistent storage should be updated also. On the next start, retained messages can then be loaded back into memory from the backing store. This covers many use cases, I think. What I did not like with the current events was, that the backing store needed to evaluate some logic to decide whether a message should be removed or added or updated (by checking the payload), hence the proposal in #1763.

Also, I assumed that, going forward, these events could be raised even when all messages are no longer kept in memory (i.e., with a SQL database behind); then the StoredRetainedMessages member in the events no longer makes sense. With a replacement of the MqttRetainedMessageManager via interface option, these events would no longer be raised, so it would not be an issue.

Maybe current events could be renamed to ...InMemoryRetainedMessagesEvents or similar as a hint that only the in-memory store will raise these events, the MqttRetainedMessageManager could be renamed to MqttRetainedMessageStoreInMemory and the interface proposed in #1663 could be renamed to IMqttRetainedMessageStore?

Alternatively, events could be removed and the MqttRetainedMessageStoreInMemory class could be moved out of the library code into an example that implements the interface. It would be slightly more effort to support "in-memory" retained messages for library users since they would need to move the example code into their application. The core library would be "cleaner" but slightly less convenient to use in this scenario. I'm leaning towards having the "in-memory" support within MQTTnet for convenience, but I'm not really sure what's best here.

The store must also support the query for matching messages based on a topic on its own. We may need to extract the topic matching code in a static or whatever class so that other "providers" can make use of it. The reason for this is that a store like SQL database may want to load messages only which are matching the topic. With the current code the storage must always return all messages so that another source makes no sense in terms of performance optimization.

Yes, what I found was, that just the FilterRetainedApplicationMessages method in the MqttClientSubscriptionsManager needed some refactoring, not a big step. In my trial, I have ended up with a slightly different approach than in #1663, but have not looked into it in detail. As far as I can see though, allowing other retained message stores to work nicely with topic matching should work without significant changes.

When we have a new store the question is who is responsible for removing expired messages? I thought just removing them when accessing the store but this would delete the messages only when there is traffic. Having a timer (or similar) which will remove the messages every second may produce high CPU load.

I have used an event driven approach for this; this is basically a server task that waits until it is time to do something (expire messages or sessions, or send a delayed will message (see https://github.com/logicaloud/MQTTnet/blob/persisted-sessions-v4/Source/MQTTnet/Server/KeyEventSchedule.cs). Although probably not strictly necessary, I have just changed it to use a tick count for event timing (instead of DateTime timestamps) to avoid issues with server time adjustments. The tick count is always 64-bit, even for systems that do not have Environment.TickCount64 (see TimeTickSource.cs in the same branch). Maybe that could become useful elsewhere where DateTime timestamps for short time spans are used.

All methods of the store must be returning a Task and accept a Cancellation Token.

Yes, agree, especially when aiming at stores where queries may take some time to complete.

Regarding the persistent sessions, this should work similar to retained messages, i.e., there could be a built-in "in-memory" option or a backing store via interface extension.

If you'd like to have a look, the proposed features are in https://github.com/logicaloud/MQTTnet/tree/persisted-sessions-v4, although all in one. It still needs some work (i.e., follow the interface approach vs events for the MqttPersistentSessionManager to be consistent with the approach decided on for the MqttRetainedMessageManager, session expiry timestamps should use ticks, potentially add a built-in "in-memory" persisted sessions option or an example, unit tests, benchmarking).

I you are happy to provide some guidance along the way, then I'd be happy to start one (or more?) pull request(s) to integrate those features.

[chkr1011]

Maybe current events could be renamed to ...InMemoryRetainedMessagesEvents or similar as a hint that only the in-memory store will raise these events, the MqttRetainedMessageManager could be renamed to MqttRetainedMessageStoreInMemory and the interface proposed in #1663 could be renamed to IMqttRetainedMessageStore?

To me this approach looks a little bit strange to me when comparing this to other SKDs like Azure Queues etc. Also from a modelling perspective. The "storage backend" is in fact a dedicated component which some responsibilities like storing and providing the messages. This would also allow to avoid storing all retained messages in memory because this is something which some people are complaining right now. They have millions of retained messages and they are all held in memory even if they are rarely used.

Alternatively, events could be removed and the MqttRetainedMessageStoreInMemory class could be moved out of the library code into an example that implements the interface. It would be slightly more effort to support "in-memory" retained messages for library users since they would need to move the example code into their application. The core library would be "cleaner" but slightly less convenient to use in this scenario. I'm leaning towards having the "in-memory" support within MQTTnet for convenience, but I'm not really sure what's best here.

We should indeed move the existing code to that new component and use this as the default instance so that users may never notice a difference (apart from all known events for this purpose are gone). But this also raises the conceptual question which component will be responsible for removing expired messages. Right now I would say that the storage implementation should be responsible because the strategy might be different (timer every 1 second or deletion when requested) depending on the actual backend and user needs.

Let us start a new branch and see where it goes. I am not available for a couple of weeks to please be patient in terms of comments and PR stuff. But I will already rethink that idea 😄

[logicaloud]

To me this approach looks a little bit strange to me when comparing this to other SKDs like Azure Queues etc. Also from a modelling perspective. The "storage backend" is in fact a dedicated component which some responsibilities like storing and providing the messages. This would also allow to avoid storing all retained messages in memory because this is something which some people are complaining right now. They have millions of retained messages and they are all held in memory even if they are rarely used.

Yes, you would need to add a Handled flag so that the in-memory storage could be skipped. Either way, and event based interface does not seem to be a good fit.

We should indeed move the existing code to that new component and use this as the default instance so that users may never notice a difference (apart from all known events for this purpose are gone). But this also raises the conceptual question which component will be responsible for removing expired messages. Right now I would say that the storage implementation should be responsible because the strategy might be different (timer every 1 second or deletion when requested) depending on the actual backend and user needs.

The notification about removal should come from MQTTnet, and how the storage backend handles it may differ (i.e., process immediately, put them in a queue for removal every second, etc.).

Let us start a new branch and see where it goes. I am not available for a couple of weeks to please be patient in terms of comments and PR stuff. But I will already rethink that idea 😄

Sounds good. I'll update a branch with the following objectives:

1. Add an IMqttRetainedMessageStore interface.
2. Derive an MqttRetainedMessageStoreInMemory class from that interface as a built-in option.
3. An external implementation for retained messages can be enabled by calling .WithRetainedMessageStore passing the IMqttRetainedMessageStore interface.
4. Follow the same approach for persistent sessions, i.e., use an IPersistentSessionStore interface with a built-in in-memory option.

I'll keep things together with the message expiry and session expiry handling for now and post here or in a pull request for review when ready.

[chkr1011]

Please let us focus on the retained messages first and then move over to the sessions store in another PR.

[logicaloud]

Sure. There is already a pull request for the IMqttRetainedMessageStore / IMqttRetainedMessageManager interface implementation (#1663). Did you want to pull this into a new branch and then go from there? Then add event based retained message expiry afterwards? So steps could be

1. Implement IMqttRetainedMessageStore interface
2. Merge into master?
3. Add event based retained message expiry
4. Merge into master?
5. Implement IMqttPersistentSessionStore and event based session expiry
6. Merge into master

[logicaloud]

BTW, the reason why I'm moving towards using the mentioned TimeTickSource instead of UTC timestamp differences for possibly short intervals (timeouts, message expiry, session expiry) is described in the discussion here: dotnet/runtime#15207

What I have observed myself is a Windows PC disconnected from the Internet adjusting its UTC clock within a month or so. The clock then jumps by a minute, which in turn can lead to timeouts and other issues. I'm not sure what other devices (i.e. single board computers) do without Internet connection. The problem is not likely to occur on servers with proper time synchronization. Either way, the problem arising from DateTime.UtcNow time jumps could be avoided by using the elapsed tick count (which, I believe, may even survive a Windows soft reboot, although I have not tested that myself).