Chat Service Architecture: Persistence

Player Base Growth

Over time, as the number of League of Legends players grew, we started encountering some trouble: the chat servers began overloading the MySQL primary instance. Players were seeing issues when loading their rosters, inviting new friends, or editing player notes. This sparked many internal discussions about the future of the MySQL backend, which led to identifying a few major pain points with the architecture as it stood:

• The MySQL primary needed to be scaled vertically—in order to achieve more capacity, we had to continually add more memory plus extremely expensive storage (e.g. FusionIO). This would have become a bottleneck to shipping value to players in the form of new features, and that’s something Riot takes extremely seriously.
• The primary also acted as a single point of failure in our system. Even small glitches in its performance resulted in timeouts when players would load their friends list. Despite having a backup server, transient problems that did not trigger failover were enough to impact the overall system. Major outages (caused by software or hardware) often snowballed into service downtime and significant manual work to resolve the issue.
• As the data set started to grow, schema migrations of any kind became extremely costly. Their application required careful planning, extreme diligence, and often hours of scheduled chat downtime. This significantly slowed development and delayed the release of features to players.

We evaluated several possible solutions to the problems listed above. Options included application-level sharding of MySQL, MySQL Cluster, Cassandra, and Riak. We encountered difficulties in quickly recovering from synchronous replication issues while evaluating both sharded and clustered MySQL. Cassandra was certainly a viable choice, but it imposed schema constraints that we really wanted to avoid. The last option - Riak - proved to be a flexible, scalable, and fault tolerant data store and we decided to proceed with it.

Riak

In the traditional world of relational database management systems (RDBMS), data is typically modelled as a set of tabular relations between entities stored in the system. In the vast majority of cases, users of RDBMS access and manage their data by means of SQL. NoSQL systems, such as Riak, depart from tabular relations. This introduces some complexity, but also a few noteworthy advantages. Typically NoSQL databases put engineers in a much more relaxed environment: schema-less interfaces allow for faster iteration; horizontal scalability removes the burden of many scale problems; and data is protected by internal replication. On a very high level, Riak is a distributed key-value NoSQL database that can be scaled horizontally in linear fashion and also provides AP (availability and partition tolerance) semantics as defined by the CAP theorem.

Migrations to Riak

Moving from MySQL to NoSQL meant major changes. We had a lot to learn about the CAP theorem, eventual consistency model, and a few other interesting concepts that Riak relies on.

To quickly rollout Riak globally, we decided to reuse our existing chat hardware and co-locate the data stores on the same servers as chat. This allowed us to avoid the long process of acquiring and rebuilding servers around the world, as well as the costs of new machines. We ended up with the following architecture:

Deploying Riak instances to the same hardware as our chat services is a simple matter of deployment tool configuration (Chef, in our case) and has proven to be a very solid and predictable way of handling our clusters.

In this architecture each chat server connects to one of the available Riak instances via a protocol buffers interface, with a preference to choose an instance running locally on the same machine. In cases when the persistent protocol buffers connection is terminated, the chat service will automatically reconnect to another instance to ensure no player data is lost. Chat servers also monitor the Riak cluster topology—whenever it changes, they try to balance the connections across all cluster members (including the locally running Riak instance).

Internally, Riak uses a leveldb bucket backend with settings that allow for multiple versions of the same object (e.g. allow_mult = true, last_write_wins = false), and a replication factor set between 3 and 5 depending on the bucket type and importance. We consider friends lists highly important and therefore replicate them to more partitions. An example of less important data could be messages sent to a player while they were not logged in.

Write conflicts arise due to the distributed nature of the system, as well as from external events such as network partitions. Aphyr provides an excellent introduction to the write contention problem on his blog. In the traditional RDBMS world, the atomicity of transactions is ensured by implementing locking with a two-phase commit protocol. This approach sacrifices availability in order to guarantee that data is always strongly consistent on all servers running in the cluster. Unfortunately, this type of consistency does not work well in a distributed ecosystem, and therefore is relaxed in favor of eventual consistency properties. We approached this by implementing application level CRDTs (Convergent Replicated Data Types, scroll to the 'State-based CRDTs' section of the Wikipedia entry) that allow chat servers to resolve potential siblings lists in a predictable and automated fashion.

Let’s consider Bob and Charlie—players who would like to become friends with Alice. It just so happens that they send friend invites to Alice at the same time, which modifies Alice’s friends list concurrently from two different chat servers. Both updates to the Alice object succeed—however, Riak internally stores both versions of the updated friends lists (one with Bob's invite, the other with Charlie's). The next time Alice logs in and chat servers read her roster from Riak they will detect the conflict and merge the logs, ending up with both invites waiting for Alice's approval.

As a result, when a chat server wants to store an object it first encodes it into a CRDT structure, which is then JSON-encoded (so it’s easily readable by multiple services), and finally compressed. Here’s an example of the structured data we store in Riak (player friends list CRDT, JSON-encoded):

[mptaszek@na1chat001 ~]$ curl localhost:8097/buckets/NA1_roster/keys/sum1234 | gunzip | jq .

{
  "v": 1,
  "log": [
    {
      "val": {
        "created_at": "2015-09-17 03:44:42",
        "priority": 2,
        "group": "**Default",
        "askm": "",
        "ask": "none",
        "sub": "none",
        "nick": "0xDEADB33F",
        "jid": "[email protected]"
      },
      "type": "add",
      "ts": 1442461482936116
    },
    {
      "val": [
        "sum98765",
        "pvp.net",
        "out"
      ],
      "type": "change_ask",
      "ts": 1442461483381002
    },
    {
      "val": [
        "sum112211",
        "pvp.net",
        ""
      ],
      "type": "del",
      "ts": 1442461487937193
    }
  ],
  "val": [
    {
      "created_at": "2015-01-02 03:12:00",
      "group": "**Default",
      "askm": "",
      "ask": "none",
      "sub": "both",
      "nick": "Teemo Ward",
      "jid": "[email protected]"
    },
    {
      "created_at": "2014-10-23 10:32:47",
      "priority": "3",
      "group": "**Default",
      "askm": "",
      "note": "Mom",
      "ask": "none",
      "sub": "both",
      "nick": "Mooooorg",
      "jid": "[email protected]"
    },
    ...
  ],
  "mod": "mod_roster_riak",
  "ts": 1442461482739733
}

As you can see in the example payload, players’ rosters are simple lists of friends, with a mutation log attached to them. This mutation log consists of all state changes in form of log statements, e.g. changing friendship status, deleting items from the list, or adding new items to the list.

Although Riak offers on-disk snappy compression we found out the hard way that large objects (1MB+) really upset it, and sibling explosion can even cause service downtime. JSON compresses extremely nicely, and allows us to reduce the 95th percentile of object sizes stored in Riak from ~100kB to ~10kB. The graph below shows the distribution of object sizes we put in Riak throughout a day in the EU West (EUW) region:

Thanks to extensive LevelDB caching and the fact that the entire social graph almost fits into memory, 99.9% of our Riak queries finish within ~8ms (we almost never hit the disk):

The loading of friends lists (including uncompressing, parsing JSON, merging CRDTs, and constructing the list in memory) takes less than 10ms in 99% of cases:

For more information, check out the rest of this series:

Part I: Protocol 
Part II: Servers
Part III: Persistence (this article)