Symfony Messenger in microservices: message contracts and resilience

Photo by Jack Anstey on Unsplash

Which libraries do you use when building communication between PHP microservices via message bus? Though this task does not sound uncommon, we ended up with custom implementation on top of Symfony Messenger component. In this article I will try to share the most important concepts of our in-house solution.

What did we try to solve?

While building our ERP system (a kind of mix of supply chain management, order processing, and accounting) we followed Spotify team model, where a team is split into smaller squads and tribes, which aim to be as autonomous as possible. This (according to Conway’s law) led to implementation of technical solution trying to reflect independence of teams, such as:

• Microservices architecture, where each service has a single owner-team.
• Continuous delivery avoiding centralized deployment trains.
• Contract-driven communication between services (and teams in turn as well).
• Prevailing of asynchronous (message bus) over synchronous (RPC/REST/etc.) communication.

In addition to the directions given above, there were few more tasks to solve regarding asynchronous communication:

• RabbitMQ was already used, so it needed to be adapted first. But in general, we wanted to be as agnostic as possible in terms of using message buses. We ended up using our solution with at least AMQP, SQS and DB transports, so the goal was achieved.
• There must be a way to enforce contracts of messages with a framework for cross-team reviews of these contracts.
• Software must be resilient.

After analysis, we came up to the conclusion that Symfony Messenger gives the most flexibility in terms of being message bus agnostic. It provides a unified wrapper, where transports (Redis, AMQP, DB, AWS SQS, etc.) are pluggable as separate dependencies.

Though there are also significant limitations of Symfony Messenger (when I write this article, Symfony 5.3 is the latest stable version):

• The component is not really adopted to work with messages published and consumed by different applications, see #33912 for example.
• There are gaps in resilience behavior, e.g. see open #36870, already resolved #36864, and a few more.

Therefore, we decided to build our custom transport (as a private composer package), which I am going to describe below.

Implementation

This transport is easy to use, one needs only to install the private package and specify a custom prefix in transport DSN instead of standard “amqp://”.

The transport was built on top of symfony/amqp-messenger, so it reuses connection classes, while sender and receiver are extended.

#1 Deserialization and contracts

Having custom transport allows introducing additional configuration options without breaking already existing options for standard “amqp-messenger”.

dsn: 'custom-amqp://...'
options:
    serialization:
        deserialize_to: 'App\Message\ProductMessage'
    validation:
        schema_path: 'product_message_schema.json'
        validate_on_consume: true

First, we needed to create a way to support multiple applications’ communication, therefore we introduced “deserialize_to” option, which tells the library which class must be used as a message object. In native Symfony Messenger deserialization happens based on message header, which contains full class name (including namespace) of initial message — which is of course hardly usable when publisher and subscriber are different applications.

Second, we wanted to enforce contract validation (we use JSON messages, therefore JSON-schema is a contract). Just having a contract is not enough, messages should be actually validated against this contract. But when should you apply it: on publish or on consume? Our reasoning was:

• Validation is obligatory on publishing, because publisher is the owner of a contract — they should avoid publishing invalid messages at any cost.
• However, validation is optional on consumption and is disabled by default.

#2 Contracts organization

All right, we have enforced teams to use contracts in messages. Now no message can be published without a contract. We all also agreed that contracts should be maintained in a backward-compatible manner: if a change is incompatible, new major version or just another contract should be introduced.

But now we have multiple messages contracts and JSON files circulating through multiple applications with unclear versioning and vague ownership and maintenance process.

In addition to that many of our JSON-schema contracts contained sub-schemas like this:

"price": {
    "description": "Promotion price",
    "$ref": "../common/money.json"
},

Where “money.json” contains an object consisting of currency and amount, which is reused and copied multiple times.

We came up with the following solution:

1. All contracts are stored in a single repository:

• This allows introducing a process of cross-team review for any contract changes, e.g. just an agreement or enforcement of merge-request reviews.
• This allows setting up notifications for any contract changes(e.g. if notifications are sent to a dedicated Slack channel).
• This solves a problem of sub-schemas, which now are described once and referenced multiple times. No copy-pasting is needed anymore.

2. Every time a change happens, contracts are automatically packed into a new version of PHP package (Symfony bundle in this case), which then can be installed/updated in publishers/consumers applications.

• This solves versioning problem: applications rely on specific contracts, which might be updated only explicitly.
• There is no limitation towards a specific technology, we might build similar package for Java/JavaScript/Python, etc.

3. Contracts also contain some metadata, e.g. description of what this message is about and destination (in terminology of RabbitMQ this is equal to “routing key”).

Additionally, such approach allowed auto-generation of documentation pages with a human-readable catalog of all data flowing through message bus in the system.

Example of messages contracts documentation

Symfony bundle containing contracts exposes parameters, which can be referenced as schema path or routing key, so no path or routing keys hardcoding is needed anymore, no manual contracts copying is needed as well.

options:
    serialization:
        deserialize_to: 'App\Message\ProductMessage'
    validation:
        schema_path: '%contracts.schema.product.path%'
    publish_routing_key: '%contracts.product.v1.routing_key%'

#3 Messages metadata

We found it very useful to inject metadata into each message, because it greatly helps with logging, debugging, and tracing.

Example of message body with metadata

This might be a subject for a separate article since the topic is not too small.

#4 Expected unexpected

Okay, what do we do when failures happen? Symfony Messenger already has a built-in retries mechanism and failure transports concept, which can be configured for consumers.

What about producers? What happens if RabbitMQ is temporary not available, or message schema validation fails? Our goal is to avoid losing data at any cost. Unfortunately, there is nothing available out of the box, so we had to implement adding to failed transport on publishing as well. Symfony Messenger has a great concept of message stamps, so we distinguish failed on publishing and failed on consuming using custom stamps added to envelopes.

It was important for us to continuously monitor failed messages (the aim is to have zero of them), because some of them might be republished/reconsumed automatically, while others need manual resolution. We monitor number of failed messages via Prometheus and Grafana, where applications expose monitoring endpoint, which contains multiple metrics including failed messages.

Finally, what should engineers do with failed messages? Symfony Messenger has CLI commands to manage them (list/view/reconsume/delete), which we extended with “republish” action. But CLI has a disadvantage as well: it means engineers should get SSH/kubectl exec/whatever else permission in production, which is kind of complicated when you work with sensitive information (e.g., customer orders info). So, we additionally implemented HTTP endpoints and UI with proper permissions restrictions for failed messages management.

See also second part: Observability with RabbitMQ and microservices