Scalability: redesign MQTT Command/response pattern

[knolleary]

Description

Part of #2782

The flow to create a Device Editor session involves the following initial steps (ref)

• An HTTP request is made to trigger the session
• An MQTT message is sent to the device to get it to connect
• The device connects its websocket
• The device published an MQTT message to confirm
• The HTTP request from the first step is completed with the session identifier.

The key point is that, in the above flow, the initial HTTP request is only responded to when a reply to the initial MQTT publish has been received over MQTT. In a scalable system, the Forge App will use a Shared Subscription to the broker - so that work can be distributed and will mean the reply could end up at a different instance.

The other place the command/response pattern is used is when the request to create a new Device Snapshot. The command is published and the device sends back the snapshot over MQTT. The original HTTP request to create the snapshot is blocked and only responds once the snapshot has been created.

Possible Solutions

1. Do not use a shared subscription for command/response MQTT pattern - and include a reply-to topic in the command message to ensure the response comes back to the topic subscribed to the correct instance.

   • Cons
     • Doesn't really solve the stateful nature of the command/response pattern
     • The current Device Agent has a hardcoded response topic to use - so we'd have a legacy issue with existing devices.

2. Redesign the whole command/response pattern to support stateless behaviour. The initial HTTP response will respond once the initial MQTT message is sent with some sort of 'pending' response including a request identifier. An API can then be used to track the completion of the pending request. The exact details of which will necessarily differ between the two current use-cases of the Command/Response pattern.

   • Pros
     • Cleanest solution that doesn't require any Device Agent changes
     • Properly stateless
   • Cons
     • Not trivial to implement properly - needs some thought on database table structure.
     • Changes the external API for settings up Device Editor and capturing Device Snapshot

Epic/Story

#2782

Have you provided an initial effort estimate for this issue?

I have provided an initial effort estimate

[knolleary]

I believe I have a design that will work with a fairly minimal set of changes (compared to the alternatives above).

Currently, the sendCommandAwaitReply function in forge/comms/devices is responsible for publishing the message and setting up the internal state to wait for the response. This is done by generating a uuid (correlationData) property that is included in the command sent to the device. The response includes this uuid so the pending promise can be found and resolved with the response.

The problem is, if we use a shared subscription on the response/device topic, the response could come to a different instance to the one holding the pending promise. Here's how I propose to solve that.

When a message is received on response/device (a shared subscription) we check the correlationData to see if it is known to this instance. If it is known, then the pending promise is resolved. That is existing behaviour. The new bit comes if the correlationData is not recognised. In that case we will do the following:

1. Write the response data to a new db table, indexed by correlationData.
2. Publish an mqtt message to response/device/broadcast (topic name tbd) with the correlationData property.
3. Crucially, this topic is subscribed to by all instances and is not a shared subscription. The instance that recognises the correlationData property can then retrieve the original message from the database (deleting it), then pass to its pending promise.

There will be a matter of tidying up the response table if a message doesn't get 'claimed' by any of the instances; we can add a housekeeper that deletes any that are more than X minutes old.

Next steps

• Create db table to hold pending response messages, including migration etc
• Add response/device/broadcast to platform ACL to allow pub/sub by the platform
• Update logic in forge/comms/devices.js as outlined above
• Create TimerTask to keep the pending response table tidy