HTTP/MQTT sync

To keep entities synchronised between B2B backend and frontend, two systems are in place.

Models that can benefit of this system should include Hairdressr::Synchable module and its behaviour must be defined in Uala::Sync module.

HTTP way

Using the /api/v1/venues/%d+/synch.json endpoint, frontend can poll backend asking about updated entities in a particular period. Logic on the front side should save last time the HTTP synch happened and poll backend every venue.api_refresh_rate seconds. This way requires the backend to handle a lot of RPM to that endpoint that means lot of database queries, so it uses a lot of resources.

MQTT way

The other way is a pub/sub connection, handled via MQTT, where backend pushes data to the frontend. In the middle of that, a MQTT broker managed by our infrastructure coordinates messages. Venues that uses this system has the boolean field active_mqtt enabled. To do that, there are helper methods in the backend that keep track of models' changes, both when Rails' callbacks are available, but also when they're skipped.

PLEASE NOTE in the Hairdressr::Synchable module, update_all method is extended.

When a change occurs, a record is stored (enqueued) inside a dedicated Redis instance with a particular format. There's then a Rails task (mqtt_sync) that process (dequeue) those messages and sent MQTT messages to the broker. To prevent sending entities update too frequently, it's required for a queue entry to stay in the queue at least for a fixed amount of time (defined using MQTT_QUEUE_MIN_TIME_SEC env var); this logic is implemented via a LUA script that runs on Redis cluster directly. Queue's entry is unique and the uniqueness is checked on enqueuing via a LUA script that runs on Redis cluster directly.

Frontend implementation requires the backend to send a ping every X seconds to know that everything is ok on its side, and to do that, another Redis queue is used to store last sent timestamp, then a LUA script check when it's time to send it. The mqtt_sync task check if it's time to send the ping, and when needed, send it and update latest sent timestamp.

MQTT flow overview

Involved components (from code POV)

1. BE (Uala Main): when any record changes in these models^ include Hairdressr::Synchable module, it pushes (aka enqueue) a formatted message into Redis (Set is used and acts as a queue, currently this Redis is a shared instance also used by other functionality). Currently we have 2 queues: venues and countries, the logic of which to use is at Hairdressr::Synchable module L20 and L138
2. Redis:
   • a temporary storage to hold these messages
   • 2 reasons for using Redis as middleman/buffer
     • to avoid data lost when MQTT Broker is unavailable (otherwise these messages would be lost)
     • to avoid spamming MQTT Broker: for example, one appointment changes might trigger many tables updated, which will spam MQTT broker. We use Redis Lua script to avoid pushing the same message (based on resource_name and its id) into Redis^^
3. BE (MQTT Sync):
   • this is a distributed^^^ Rails rake task runs infinitely, to dequeue messages from Redis then publish these into MQTT broker
   • it pings server (publish MQTT_PING_TOPIC to MQTT broker) every X seconds, that is the topic FE uses to understand if MQTT broker is still alive
   • it also sends metrics to Prometheus (and dashboard in Grafana)
4. MQTT broker: standard MQTT broker
5. FE (Uala Web b2b): Connects to MQTT broker to receive MQTT messages on subscribed MQTT topics

Flowchart

Annotations

• ^: Hairdressr::Synchable module currently included in 41 models/tables
  • appointment.rb
  • billing_company.rb
  • checkout.rb
  • custom_time_table.rb
  • customer.rb
  • customer_cart.rb
  • customer_promotion.rb
  • customer_tag.rb
  • deadline_typology.rb
  • debt.rb
  • extra_opening.rb
  • fiscal_day.rb
  • line_item.rb
  • manufacturer.rb
  • marketing_promotion.rb
  • marketing_promotion_points_transaction.rb
  • newsletter.rb
  • notification_credit.rb
  • order.rb
  • order_delivery_note.rb
  • package.rb
  • payment_method.rb
  • product_category.rb
  • product_line.rb
  • promotion.rb
  • staff_member.rb
  • staff_member_treatment.rb
  • supplier.rb
  • transaction.rb
  • treatment.rb
  • vacancy.rb
  • venue.rb
  • venue_action.rb
  • venue_cost.rb
  • venue_cost_item_typology.rb
  • venue_product.rb
  • venue_service_good.rb
  • venue_treatment.rb
  • venue_treatment_group.rb
  • workstation.rb
  • workstation_treatment.rb
• ^^: It uses SSCAN to find exist string but doesn't works because that command only offer limited guarantees. So when the set has too many elements, there is a possibility to push the same message again due to uniqueness check failure.
• ^^^: 12 instances/pods running by default (config in uala-iac - 10 heavy and 2 light)

Useful links

1. TWPRO Load Balance dashboard to see overall healthy of pods
2. Sematext MQTT incident dashboard: contains most of info when something went wrong, most important one is response time for both Italy and EU pods
3. Grafana TW Backend(VPN required): to monitor trends of how many messages in Redis waiting to be dequeued
4. Prometheus(VPN required): real-time metric of messages in Redis

References

1. TWPRO System architecture overview Miro 2023-11-21 by Rizzo & Panda
2. TWPRO System architecture overview recording 2023-11-21 by Rizzo & Panda
   • MQTT relate key points (in Bank's opinion):
     • 27:40 ~ 28:42: Luca clarifies communication among BE, FE, MQTT sync and MQTT broker from infrastructure POV
     • 39:53 ~ : Data sync in general deep dive starts
     • 41:40 ~ : from FE POV, if browser pauses the tab, MQTT is also paused. So when a tab becomes active again, FE calls synch to retrieve messages happen during the pause
     • 42:20 ~ : Panda explains why HTTP sync still exist (for overnight data changes)
     • 47:56 ~ : Rizzo explains FE fallback strategy - BE publish a ping message every 5 seconds into the MQTT_PING_TOPIC that all clients subscribe to, to determine if no data in MQTT because of no data or MQTT is down. When FE cannot receive updates from MQTT_PING_TOPIC, it fallback to HTTP sync strategy
     • 49:30 ~ : Q: why migrations sometime cause problems? A: update_all is monkey-patched and will enqueue a lot if called, which could cause problems
     • 53:30 ~ : Q: what would happen if updated_all is triggered? A: MQTT queue would be full and it might take 30mins to process, so no real-time data in FE in these 30 mins
     • 55:45 ~ : Why would venue have MQTT disable? A: It should not. There were some due to testing purpose. AA has a switch to control if venue is using MQTT or not, but Panda sees no reasons to disable it. There were some venues' firewall block non-http traffic but we use normal 443 port so it's always enable

MQTT Authentication / Authorisation improvement spike (TWPRO-9666)

Original from comments of TWPRO-9666 MQTT Authentication/authentication has probably room of improvement from performance side

How MQTT Authentication / Authorisation work

1. PRO uses EMQX, an open source MQTT broker at version 4.2.1, which allows to customise authentication and authorisations in MQTT broker level
2. Production EMQX settings are in uala-iac/applications/environments/twpro-prod/envs/backend-main-mqtt-broker-vars.env
   • these settings define 3 MQTT specific endpoints and its payload format
     • authentication
       • mqtt/sessions: RPM 500~1K (general users)
       • mqtt/super_user: RPM 500~1K (super users)
     • authorisation
       • mqtt/acl_authorization: RPM 1.5K~2.5K
3. K8s currently have dynamic number of web-b2b-mqtt pods to handle MQTT Authentication / Authorisation requests
   • HPA defines 4 ~ 12 pods, depends on requests (details are in config)
4. These endpoint are called by MQTT broker when it receives MQTT client (mostly our FE) requests to connect MQTT broker and subscribe to MQTT topic

Potential improvements:

1. Investigate and optimise HPA max pods during peak time as running 12 pods to handle such small amount of request is a bit waste (saving cost)
   • looks like a quick win, but need to do it carefully, ideally pair with Platform team
2. Reduce authentication/authorisation requests
   • removing it
     • we agreed to keep as it is for security reason
   • changing it
     • it needs digging deeper into its logic, which seems not worth at the moment due to lower priority
3. Adding a delay in FE when trying to reconnect due to disconnected during MQTT outage (either no ping or MQTT broker is down, in our case is formal one) to avoid overwhelming these endpoints/pods
   • we don’t want to do this due to UX
     • Once connection is re-established, we want users to get up-to-date data immediately.
4. replace MQTT with other simpler technology
   • probably not worth doing it due to lower priority