There are a number of properties that are non-negotiable in any software system, such as no data loss, no stalling, and timely recovery from faults. These properties are particularly difficult to achieve in a distributed system, as the vast number of permutations of timings, event orderings, and faults makes reasoning about a distributed system extremely challenging.

If we’ve learned one thing developing Rama, it’s this: a distributed system is only as robust as the quality and thoroughness of its tests. More so, there are certain testing techniques which must be used if you’re going to have any hope of building a robust system resilient to the crazy edge cases that inevitably happen in production. If a distributed system isn’t tested with these techniques, you shouldn’t trust it.

Rama being such a general purpose system, capable of handling all the computation and storage needs for entire backends at scale, has a particularly large testing surface. First, there’s the core functionality of Rama to test: distributed logs (depots), ETL topologies (streaming and microbatching), indexes (PStates), and queries. Then there are the “module transition” operations for updating or scaling running Rama applications. Finally, there’s replication. Replication needs to be tested when healthy, when all replicas are keeping up, and also when replicas fall behind and need to catch up. Compounding the testing surface is that all of these need to be tested under the conditions of any number of faults happening at any time, such as disk errors, nodes losing power, and network partitions.

Testing is largely a sampling problem. Each sample exercises the system at a particular state, with input data of some size and shape, at some amount of load, and with some set of faults at some frequency. A testing strategy needs to sample this input space in a representative way. In a highly concurrent distributed system, where there are so many ways that events can be randomly ordered across different threads, achieving a representative sample is difficult. And if something isn’t tested, it’s either broken or will be broken in the near future.

Another huge issue in testing is reproducibility. A test that’s hard to reproduce is very expensive, or possibly even impossible, to debug. Test failures in concurrent systems are frequently difficult to reproduce due to the inherent randomness of its execution. As you’ll see shortly, this is another major focus of our testing strategy for Rama.

To fully cover the testing surface of Rama, we use multiple strategies. The first line of defense is our unit tests, which we run twice every hour as well as with every commit. In addition to that, we do automated chaos testing 24/7 on multiple distributed clusters. The last major piece is load testing, where we run large clusters for long periods of time with large amounts of data. There’s lots of detail to how we approach all of these which I’ll cover in the subsequent sections.

The straightforward way to unit test is to run the system like how it would be operating in production, with many threads and lots of concurrency. In this context, instead of running the system across many processes and many nodes, you instead simulate the various processes comprising the system all within the same testing process. Intuitively, running a system as close as possible to how it operates in production seems like a good way to test. However, this approach has massive flaws that make it a poor basis for which to unit test a distributed system.

In Rama we implemented this approach in a testing tool called “in-process cluster” (IPC). For basic things, like verifying how PStates are updated in a module, this approach works fine. Failures in tests like these are fairly easy to track down. However, when testing more complex scenarios like leadership switches during processing, random worker deaths, or far-horizon catchup, debugging can get extremely difficult.

When developing tests for scenarios like those using IPC, we frequently ran into situations where we would get a test failure that would only reproduce once in a hundred test runs. Sometimes it would be even less, like one in a thousand test runs. To debug the test, we would add logging and then run the test over and over until it reproduced. Unfortunately, in IPC the very act of adding logging changes the timings of execution and can make reproduction less likely. It may make reproduction so unlikely that it’s basically impossible to reproduce. So then you have to get very creative to find ways to get more information without changing the timing so much that it no longer reproduces. This cycle of adding logging and reproducing is very time-consuming. Sometimes a single bug would take us weeks to figure out.

The expense of debugging isn’t the worst issue of IPC though. The worst issue is how difficult it is to thoroughly explore the testing space. The vast majority of issues that we’ve debugged in Rama have had to do with ordering of events. Many bugs can be triggered by one particular thread getting randomly stalled for an unusual amount of time (e.g. from GC). Other bugs can come from rare orderings of events on a single thread. Exploring these scenarios in IPC requires using semaphores to try to coordinate the threads in question into these unusual permutations, or just hoping that random timings during execution sometimes explore these orderings. Exploring internal timeouts and the consequences of those requires even more creativity.

Fortunately, we found a better way.

Inspired by what the FoundationDB team wrote about simulation testing, we invested heavily into implementing “deterministic simulation” for Rama. Deterministic simulation is counterintuitive, but it is by far the best way to unit test a distributed system.

The idea of deterministic simulation is to run the whole system on a single thread. What would normally be its own thread in production is instead an executor explicitly managed by a central controller. The controller runs in a loop choosing which executor gets to run an event each iteration. A random number generator is used to choose executors, and a test is fully reproducible by just using the same seed for the RNG. Tests are easily debugged since adding logging does not change the reproducibility of the test like it frequently does for IPC.

Deterministic simulation removes all concurrency from execution of Rama during tests. This seems like it would be a bad thing by making the unit test environment fundamentally different from production. However, our experience that the vast majority of issues have to do with event ordering and timing means the exact opposite. Deterministic simulation is incredible – almost magical – for diagnosing and debugging these issues. Deterministic simulation isn’t sufficient as a complete testing strategy, as you still need tests that exercise potential concurrency issues, but it is overwhelmingly better for most tests.

Implementing deterministic simulation required us to refactor the Rama implementation to be able to run in either “production mode” using many threads, or “simulation mode” using a single thread. We also needed to abstract the notion of time. In simulation, all entities interact with a time source that’s managed by the central controller of simulation. Usually, we advance time by a small amount per iteration, but sometimes we use other strategies. For examples, tests specifically of timeouts usually disable automatic time advancement and advance time explicitly. They may advance time to one millisecond before the timeout, verify nothing happens, and then verify the timeout behavior is triggered when advancing by one additional millisecond.

Here’s pseudocode on what a simulation test looks like:

Besides reproducibility, deterministic simulation hugely improves our ability to explore the testing space of Rama. We are no longer beholden to random timings or require burdensome coordination with semaphores to explore unusual event orderings. We can suspend a particular executor to simulate a GC pause. We can use a non-uniform distribution to prefer certain executors over others. We can insert faults or other disturbances at extremely precise moments.

Each run of a simulated test is a sample of the testing space, so it’s critical to continually run all tests all the time. Each seed explores a different permutation of events. We run our full build twice an hour as well as with each commit, so we get at least 48 samples of each test every day.

Many of our simulation tests check specific behaviors, like “when a microbatch fails, the work is retried from the previous commit”. A different category of simulation tests are what we call “uncoordinated simulation tests”.

In an uncoordinated simulation test, we create many entities that interact with a Rama cluster with one or more modules deployed on it. Some of those entities may be appending new data. Others may be querying PStates. Others may be performing module operations like module update or scaling. Then there may be some antagonistic entities disturbing the cluster, like creating random disk faults, killing workers at random, or suspending executors at random. Each entity has a rate at which they perform these actions, such as “one depot append every 100 milliseconds of simulated time”. Most importantly, the entities are not coordinated with one another and take actions independently. They’re registered as executors in the simulation and are given opportunity to run events when chosen by the central controller. In tests like these, we verify high-level properties like no data loss, no stalling, data being processed in the correct order, and all replicas eventually catching up to become part of the ISR.

Uncoordinated simulation tests are particularly good at finding race conditions. The randomness and lack of coordination causes runs of the test to eventually explore all possible race conditions. And since the test is fully reproducible, we can easily track down the cause of any failures no matter how obscure the event ordering.

As mentioned, deterministic simulation is only part of our testing strategy. As powerful as it is, there are a few critical things it doesn’t cover:

• True concurrency issues, like improper concurrent access to shared memory
• Memory or other resource leaks
• Running cluster as separate processes on separate nodes

To cover these gaps, another critical piece is automated testing on a few distributed clusters that we run 24/7. We call these our “quality clusters”.

Each cluster has a dedicated process controlling it. This central process interacts with the cluster either by: appending new data, doing queries, or performing a variety of disturbances. The disturbances include random worker kills, full cluster restarts, network partitions, and suspending nodes to force them into far-horizon catchup. The central process performs high-level assertions about no data loss, cluster integrity, nothing being stalled, and nothing unexpected in any log file.

There’s tension between disturbing the cluster, which often involves killing/restarting processes, and also wanting to test long-lived processes to verify there’s no memory leaks. So the central process chooses one task group and any workers containing replicas of that task group to be protected for the lifetime of the cluster. Any routines that cause process death never pick those workers. This enables the cluster to test long-lived processes as well as the full range of disturbances.

We have three quality clusters, called “nightly”, “disturbances-monthly”, and “module-operations”. The “nightly” cluster is replaced every day with the latest Rama commit and is meant to catch recent regressions quickly.

The “disturbances-monthly” cluster is replaced every month with the latest commit. By the end of the run, if all went well, the longest-lived process will have been running for a month. This cluster deploys many modules exercising the full feature-set of Rama, and it also runs a special module called “LoadModule” which creates heavy load on all the other modules. “disturbances-monthly” performs the full range of disturbances. It does not do any module operations like module update or scaling, however, since this cluster tests long-lived processes and those routines replace every process for a module.

The “module-operations” cluster runs the same set of modules as “disturbances-monthly” and is dedicated to exercising module update and scaling. It also performs disturbances during these module operations to verify their fault-tolerance. A module operation should always either succeed completely or abort. An abort can be due to there being too much chaos on the cluster, like such frequent worker kills that the operation can’t go through in a reasonable amount of time. “module-operations” verifies these operations never stall under any conditions and that there’s never any data loss.

Issues surfaced by these clusters are much harder to debug as they are frequently difficult to reproduce. If the cause of an issue isn’t obvious, our first line of attack is to try to reproduce the issue in a simulation test. If it reproduces in simulation, then it’s easy to debug from there. This has proven to be effective for most issues surfaced, and then that simulation test serves as a regression test moving forward.

Another type of testing we do is load testing. The goal of load testing is to test bigger clusters with bigger individual partitions than we achieve in our month-long quality clusters. When everything is bigger, things like timeouts and stall detection are better exercised.

Load testing is expensive, requiring us to run hundreds of nodes for long periods of time. So we currently do load tests as one-offs rather than continuously (e.g. ahead of a release). For example, leading up to the launch of our Twitter-scale Mastodon instance in August, we did thorough load testing for long periods of time.

We’re currently working on implementing incremental backups for Rama. Once that’s done, we’re planning to use that in our quality clusters to start off each new cluster with the data that accumulated on the last cluster. This will enable our quality testing to operate with large individual partitions, which will enable continuous testing of that one aspect of load testing.

We take testing very seriously at Red Planet Labs. We spend more than 50% of our time on testing, and we believe any software team building infrastructure must do the same to be able to deliver a robust tool.

We’re constantly working on improving our test coverage, like by incorporating new kinds of faults in our simulations and quality clusters. An idea we had recently is “long-running simulation tests” where we run simulations for days or weeks at a time. This testing strategy is interesting, since though any failures will be reproducible, a failure that happens after two weeks of running will still take two weeks to reproduce. But possibly these simulations will be able uncover issues that only occur after extended runtimes, and being able to reproduce a potentially obscure issue is still very valuable.

Software cannot be understood purely in the abstract. It requires empirical evidence to know how it behaves in the strenuous conditions it will face in real-world deployments. A major reason it took us 10 years to build Rama was going through that process of testing, iterating, and testing some more until we were confident Rama was ready for production use.

Today we’ve released Rama’s Clojure API, including detailed reference documentation and API docs. Information about how to add it as a dependency in your projects is on this page.

Rama is a new programming platform for building high-performance scalable backends, integrating and generalizing data ingestion, processing, indexing, and querying. It’s so general-purpose that it can build entire backends with extremely diverse computation and storage needs on its own, without any of the impedance mismatches that have plagued backend development for decades. One way to think of Rama is as a “programmable datastore on steroids”, where you mold your datastore to fit your application rather than the other way around. It can build interactive, consumer-facing applications just as easily as it can build complex analytics applications.

In August we revealed Rama for the first time by building and operating a Twitter-scale Mastodon instance that’s 100x less code than Twitter wrote to build the equivalent. We ran the instance with 100M bots posting 3,500 times per second at 403 average fanout. In the blog post, we explored the implementation of this instance in depth and how it makes use of Rama’s unique capabilities. The code for that instance is open source.

This post is a self-contained introduction to Rama and its Clojure API. The Clojure API is actually the native API to Rama, and the Java API we released in August is a thin wrapper around it. I’ll start the exploration with a brief overview of the concepts of Rama, and then I’ll dive into the ins and outs of using the Clojure API.

Every backend must balance what information gets precomputed versus what gets computed on the fly at query time. As a developer, you must tackle four things when building a backend: how to receive new data, what indexes to create with what structures, how to process new data, and how to query your indexes to serve application requests.

A typical architecture might look like this:

• An API server receives new data
• The API server writes that data, potentially in aggregate form, into one or more databases (e.g. Postgres, ElasticSearch, Cassandra, etc.)
• The API server handles application requests by querying one or more databases

Sometimes a queue system is inserted between the API server and the backend, with a separate system deployed (e.g. custom workers, Kafka Streams, Storm) to process that queue and update the databases. There are tons of variations of what combinations of tools are used to construct a backend, and there’s further variation in what tools are used to deploy and monitor these systems.

There’s many performance, scalability, and fault-tolerance issues to tackle when building systems this way. But the most insidious problem is complexity. Just the sheer number of tools you have to operate creates huge integration and deployment complexity. Plus, each tool you use is narrow and only able to handle certain use cases. With databases, you frequently have to twist your domain model to fit the database’s data model, creating major impedance mismatches at the core of your backend.

Rama can build entire backends end-to-end, integrating and generalizing every aspect of what it takes to do so. Applications built with Rama are scalable, high-performance, and fault-tolerant. Rama has four concepts, corresponding to “data ingestion”, “data processing”, “indexing”, and “querying”:

Rama gives you total flexibility in designing what gets precomputed in your backend versus what gets computed at query time. At a high-level, this programming model is event sourcing with materialized views. When programming Rama, you materialize as many views as needed in whatever shapes are most optimal to serve your application’s use cases.

Let’s start with how indexes work in Rama, as this is a key part of how Rama is so flexible and reduces complexity so much. Indexes are called “partitioned states”, which we usually refer to as “PStates”. Unlike a database, which has a strict “data model” (e.g. relational, key/value, document, column-oriented, graph, etc.), PStates are specified in terms of data structures. Each “data model” is really just a specific combination of data structures – key/value is a map, document is a map of maps, column-oriented is a map of sorted maps, and so on. PStates allow you to express all those data models using the simpler primitive of data structures, and it can express infinite more “data models” by combining data structures in other ways.

PStates are durable, distributed, and replicated. That is, they’re not in-memory structures, and each partition can be much larger than the memory on a machine. Even nested data structures can be larger than memory, and you can still read and write to them extremely efficiently.

Data comes in to Rama through “depots”. A “depot” is a durable, distributed, and replicated log of data. They’re similar to Apache Kafka except integrated with the rest of Rama.

"ETL"s, extract-transform-load topologies, process incoming data from depots as it arrives to materialize PStates. Most of the time spent programming Rama is spent making ETLs. As you’ll see shortly, the ETL API is extremely expressive. It’s a Turing-complete dataflow-based API that seamlessly distributes computation.

The last concept in Rama is “query”. As will be no surprise to Clojure programmers hearing the description of PStates as arbitrary combinations of data structures, Specter’s paths are the core API for querying them. Rama has an internal fork of Specter that adds powerful reactive capabilities to them. In addition to this, you can also make “query topologies”. These are predefined, on-demand, realtime, distributed queries that can query and aggregate data across any number of PStates and any number of the partitions of those PStates. These are the analogue of “predefined queries” in traditional databases, except programmed with the same dataflow API as used to program ETLs and far more capable.

All these concepts are packaged together into a Rama application as a “module”. A module is an arbitrary collection of depots, ETLs, PStates, and query topologies. Modules are launched onto a Rama cluster, and they can later be updated with new code or scaled up/down.

For examples of different ways in which these concepts are combined towards extremely different use cases, you can read about our Mastodon implementation or check out the self-contained, thoroughly commented examples in the rama-demo-gallery project.

We’ll start with a basic word count application, and then we’ll build an auction application with timed listings, bids, and notifications of winners and losers. As we go, I’ll explain the various pieces of the Clojure API.

First, let’s require the namespaces needed:

com.rpl.rama.path is Rama’s internal fork of Specter, and for the most part it’s API-equivalent to the open-source version.

Next, let’s define the word count module:

Partitioners are a great example of how seamless it is to write distributed code with Rama. Because they’re based on the same “call and emit” paradigm as all other operations, code that’s moving around the cluster like this can be read linearly. And because they’re no different from other operations, they compose with other dataflow code just like any operation. As you’ll see in the next section, you can also express conditionals and loops with the dataflow API. So you can trivially do things like a looping computation with partitioners in the body that hops around the cluster with each iteration of the loop.

Next, let’s launch the module on it. “Tasks” are Rama’s name for a module’s partitions and refer to the fact that partitions of a module perform both computation and storage. Here we run four tasks across two threads:

Next, let’s fetch clients to the depot and PState of the module:

The term “foreign” refers to Rama objects that live outside of modules. Now, let’s append some data to the depot:

Queries on PState clients use paths to express the query. These examples are extremely simple since we’re just fetching the values in a map, but you’ll see more complicated queries in the auctions example later in this post.

That’s all there is to it. The way in which depot and PState clients are fetched and used with IPC is exactly the same as you would interact with a real cluster.

While these examples used Rama’s blocking API, it’s also important to note there are non-blocking variants of all foreign methods that return CompletableFuture objects. These include foreign-append-async!, foreign-select-one-async, and others.

Before building the auction application, let’s briefly explore Rama’s dataflow API. As described, operations in this API are based on “call and emit” rather than the “call and response” you’re used to from Clojure and most other languages.

You can explore the dataflow API from the REPL outside the context of modules. The only parts of Rama not available in this context are partitioners since they don’t make sense in this single-threaded context. Let’s start by printing “Hello, world!”:

This is only a taste of Rama dataflow, and there’s a lot more to explore. Fragments being a generalization of functions are a very potent concept, and much of Rama’s implementation is written in this language. Fragments and dataflow are excellent abstractions for writing parallel, asynchronous, and reactive code, which is why they’re the basis of ETLs and query topologies. It’s also worth noting that Rama dataflow compiles to very efficient bytecode. For more information about the dataflow API, check out this page from the documentation.

Let’s take a look at a slightly larger example that showcases more of what Rama can do. We’ll build an auction application with timed listings, bids, and notifications of winners and losers. This application will utilize multiple PStates and demonstrate the advantages of colocating computation and storage.

First, let’s do the necessary requires:

Let’s build up the module step by step, starting with making and viewing listings. Then we’ll add bids and notifications afterward.

Here’s the data type we’ll use to represent a listing:

Next, let’s define the depot to receive new listings:

Let’s run a quick test to verify it works:

Now, let’s add bids to the module. In the next section we’ll finish the module by adding expirations and notifications. Adding bids will require two new records:

The first record represents a new bid on a listing, and the second record will be used in one of the new PStates.

Next, let’s define the depots for the module:

Adding bids doesn’t change anything about processing listings, so this part is exactly the same as the previous section. Next, let’s add the logic to process bids:

Let’s run another quick test to verify this works:

Running this prints:

Let’s finish the application by adding listing expirations and notifications. Let’s start by defining some helper functions needed by the implementation:

We’ll also need one new record definition:

A separate topology will be handling notifications, and this record will be needed by that topology.

Next, let’s once again define the depots for the module:

This is exactly the same as before. Next, let’s begin defining the ETL:

This defines a microbatch topology. Whereas streaming processes data immediately, microbatching processes a small batch of data across all depot partitions at the same time. Each iteration of microbatching processes the data that accumulated the last microbatch iteration. Since there’s no per-record overhead, microbatching has even higher throughput than streaming. However, because it’s batch-based the processing latency of microbatching is a few hundred milliseconds as opposed to the one or two milliseconds of streaming. Microbatching also provides exactly-once processing semantics for updates to PStates, even if a machine explodes in the middle of processing and the microbatch is retried.

Because they’re part of the same module, this new microbatch ETL is colocated with the streaming ETL and all its PStates. They share the same resources and can read each other’s PStates directly.

Finally, let’s run another quick test to verify it works:

This prints:

That’s all there is to it. In just 100 lines of code we’ve built a high-performance auction application that can scale to millions of listings/bids per second, is completely fault-tolerant, and is easy to evolve over time with new features. Since deployment and monitoring are built-in to Rama, this is production-ready. We didn’t implement account registration or profiles, but that’s trivial to add.

Rama derives its power from being based on composable abstractions. PStates are a composable abstraction for storage, enabling any database’s data model (plus infinite more) to be expressed as the composition of data structures. Rama’s dataflow API is a composable abstraction for distributed computation, enabling you to seamlessly combine regular logic with partitioners, yields, and other asynchronous tasks.

Rama can handle all the computation and storage for a backend, but it’s also easy to integrate with existing architectures. Rama’s integration API allows you to use external databases, queues, monitoring systems, or other tools with your modules.

Besides the documentation, we’ve released other resources for learning Rama. rama-demo-gallery contains short, self-contained, thoroughly commented examples of using Rama to build a variety of use cases. But the best way to learn Rama is to try it out yourself using the publicly available build. The REPL is an invaluable environment for experimenting with Rama. If you have any questions, feel free to ask on the rama-user Google group or #rama channel on Clojurians.

Finally, if you’d like to use Rama in production to build new features, scale your existing systems, or simplify your infrastructure, you can apply to our private beta. We’re working closely with each private beta user to not only help them learn Rama, but also actively helping code, optimize, and test.

I’m going to cover a lot of ground in this post, so here’s the TLDR:

• We built a Twitter-scale Mastodon instance from scratch in only 10k lines of code. This is 100x less code than the ~1M lines Twitter wrote to build and scale their original consumer product, which is very similar to Mastodon. Our instance is located at https://mastodon.redplanetlabs.com and open for anyone to use. The instance has 100M bots posting 3,500 times per second at 403 average fanout to demonstrate its scale.
• Our implementation is built on top of a new platform called Rama that we at Red Planet Labs have developed over the past 10 years. This is the first time we’re talking about Rama publicly. Rama unifies computation and storage into a coherent model capable of building end-to-end backends at any scale in 100x less code than otherwise. Rama integrates and generalizes data ingestion, processing, indexing, and querying. Rama is a generic platform for building application backends, not just for social networks, and is programmed with a pure Java API. I will be exploring Rama in this post through the example of our Mastodon implementation.
• We spent nine person-months building our scalable Mastodon instance. Twitter spent ~200 person-years to build and scale their original consumer product, and Instagram spent ~25 person-years building Threads, a recently launched Twitter competitor. In their effort Instagram was able to leverage infrastructure already powering similar products.
• Our scalable Mastodon implementation is also significantly less code than Mastodon’s official implementation, which cannot scale anywhere near Twitter-scale.
• In one week we will release a version of Rama that anyone can download and use. This version simulates Rama clusters within a single process and can be used to explore the full Rama API and build complete prototypes. We will also release the Rama documentation at that time. (Instructions for downloading Rama are now available here and the documentation is available here.)
• In two weeks we will fully open-source our Mastodon implementation (This is now open-source).
• Red Planet Labs will be starting a private beta soon to give companies access to the full version of Rama. We will release more details on the private beta later, but companies can apply here in the meantime.

We recognize the way we’re introducing Rama is unusual. We felt that since the 100x cost reduction claim sounds so unbelievable, it wouldn’t do Rama justice to introduce it in the abstract. So we took it upon ourselves to directly demonstrate Rama’s 100x cost reduction by replicating a full application at scale in all its detail.

First off, we make no comment about whether Mastodon should be scalable. There are good reasons to limit the size of an individual Mastodon instance. It is our belief, however, that such decisions should be product decisions and not forced by technical limitations. What we are demonstrating with our scalable Mastodon instance is that building a complex application at scale doesn’t have to be a costly endeavor and can instead be easily built and managed by individuals or small teams. There’s no reason the tooling you use to most quickly build your prototype should be different from what you use to build your application at scale.

Our Mastodon instance is hosted at https://mastodon.redplanetlabs.com. We’ve implemented every feature of Mastodon from scratch, including:

• Home timelines, account timelines, local timeline, federated timeline
• Follow / unfollow
• Post / delete statuses
• Lists
• Boosts / favorites / bookmarks
• Personalized follow suggestions
• Hashtag timelines
• Featured hashtags
• Notifications
• Blocking / muting
• Conversations / direct messages
• Filters
• View followers / following in order (paginated)
• Polls
• Trending hashtags and links
• Search (status / people / hashtags)
• Profiles
• Image/video attachments
• Scheduled statuses
• ActivityPub API to integrate with other Mastodon instances

There’s huge variety between these features, and they require very different kinds of implementations for how computations are done and how indexes are structured. Of course, every single aspect of our Mastodon implementation is scalable.

To demonstrate the scale of our instance, we’re also running 100M bot accounts which continuously post statuses (Mastodon’s analogue of a “tweet”), replies, boosts (“retweet”), and favorites. 3,500 statuses are posted per second, the average number of followers for each post is 403, and the largest account has over 22M followers. As a comparison, Twitter serves 7,000 tweets per second at 700 average fanout (according to the numbers I could find). With the click of a button we can scale our instance up to handle that load or much larger – it would just cost us more money in server costs. We used the OpenAI API to generate 50,000 statuses for the bots to choose from at random.

Since our instance is just meant to demonstrate Rama and costs money to run, we’re not planning to keep it running for that long. So we don’t recommend using this instance for a primary Mastodon account.

One feature of Mastodon that needed tweaking because of our high rate of new statuses was global timelines, as it doesn’t make sense to flood the UI with thousands of new statuses per second. So for that feature we instead show a small sample of all the statuses on the platform.

The implementation of our instance looks like this:

The Mastodon backend is implemented as Rama modules (explained later on this page), which handles all the data processing, data indexing, and most of the product logic. On top of that is our implementation of the Mastodon API using Spring/Reactor. For the most part, the API implementation just handles HTTP requests with simple calls to the Rama modules and serves responses as JSON. We use Soapbox to serve the frontend since it’s built entirely on top of the Mastodon API.

S3 is used only for serving pictures and videos. Though we could serve those from Rama, static content like that is better served via a CDN. So we chose to use S3 to mimic that sort of architecture. All other storage is handled by the Rama modules.

Our implementation totals 10k lines of code, about half of which is the Rama modules and half of which is the API server. We will be fully open-sourcing the implementation in two weeks.

Our implementation is a big reduction in code compared to the official Mastodon implementation, which is built with Ruby on Rails. That codebase doesn’t always have a clear distinction between frontend and backend code, but just adding up the code for clearcut backend portions (models, workers, services, API controllers, ActivityPub) totals 18k lines of Ruby code. That doesn’t include any of the database schema definition code, configurations needed to run Postgres and Redis, or other controller code, so the true line count for the official Mastodon backend is higher than that. And unlike our Rama implementation, it can’t achieve anywhere near Twitter-scale.

This isn’t a criticism of the Mastodon codebase – building products with existing technologies is just plain expensive. The reason we’ve worked on Rama for so many years is to enable developers to build applications much faster, with much greater quality, and to never have to worry about scaling ever again.

Here’s a chart showing the scalability of the most intensive part of Mastodon, processing statuses:

As you can see, increasing the number of nodes increases the statuses/second that can be processed. Most importantly, the relationship is linear. Processing statuses is so intensive because of fanout – if you have 15M followers, each of your statuses has to be written to 15M timelines. Each status on our instance is written to an average of 403 timelines (plus additional work to handle replies, lists, and conversations).

Twitter operates at 7,000 tweets per second at 700 average fanout, which is equivalent to about 12,200 tweets / second at 403 average fanout. So you can see we tested our Mastodon instance well above Twitter-scale.

Here’s a chart showing the latency distribution for the time from a status being posted to it being available on follower timelines:

These numbers are a bit better than Twitter’s numbers. Because of how unbalanced the social graph is, getting performance this good and this reliable is not easy. For example, when someone with 20M followers posts a status, that creates a huge burst of load which could delay other statuses from fanning out. How we handled this is described more below.

Lastly, here’s a chart showing the latency distribution for fetching the data to render a user’s home timeline:

Rendering a home timeline requires a lot of data from the backend: a page of statuses to render that aren’t muted/filtered, stats on each status (number of replies, boosts, and favorites), as well as information on the accounts that posted each status (username, display name, profile pic). Getting all this done in an average of 87ms is extremely efficient and a result of Rama being such an integrated system.

The numbers I’ve shared here should be hard to believe: a Twitter-scale Mastodon implementation with extremely strong performance numbers in only 10k lines of code, which is less code than Mastodon’s current backend implementation and 100x less code than Twitter’s scalable implementation of a very similar product? How is it possible that we’ve reduced the cost of building scalable applications by multiple orders of magnitude?

You can begin to understand this by starting with a simple observation: you can describe Mastodon (or Twitter, Reddit, Slack, Gmail, Uber, etc.) in total detail in a matter of hours. It has profiles, follows, timelines, statuses, replies, boosts, hashtags, search, follow suggestions, and so on. It doesn’t take that long to describe all the actions you can take on Mastodon and what those actions do. So the real question you should be asking is: given that software is entirely abstraction and automation, why does it take so long to build something you can describe in hours?

At its core Rama is a coherent set of abstractions for expressing backends end-to-end. All the intricacies of an application backend can be expressed in code that’s much closer to how you describe the application at a high level. Rama’s abstractions allow you to sidestep the mountains of complexity that blow up the cost of existing applications so much. So not only is Rama inherently scalable and fault-tolerant, it’s also far less work to build a backend with Rama than any other technology.

Let’s now dive into Rama. We’ll start with a high-level overview of Rama’s concepts. Then we’ll look at how some of the most important parts of our Mastodon instance are implemented in terms of these concepts. Finally, we’ll look at some code from our Mastodon implementation.

Rama is programmed entirely with a Java API, and Rama’s programming model has four main concepts:

On the left are “depots”, which are distributed, durable, and replicated logs of data. All data coming into Rama comes in through depot appends. Depots are like Apache Kafka except integrated with the rest of Rama.

Next are "ETL"s, extract-transform-load topologies. These process incoming data from depots as it arrives and produce indexed stores called “partitioned states”. Rama offers two types of ETL, streaming and microbatching, which have different performance characteristics. Most of the time spent programming Rama is spent making ETLs. Rama exposes a Java dataflow API for coding topologies that is extremely expressive.

Next are “partitioned states”, which we usually call “PStates”. PStates are how data is indexed in Rama, and just like depots they’re partitioned across many nodes, durable, and replicated. PStates are one of the keys to how Rama is such a general-purpose system. Unlike existing databases, which have rigid indexing models (e.g. “key-value”, “relational”, “column-oriented”, “document”, “graph”, etc.), PStates have a flexible indexing model. In fact, they have an indexing model already familiar to every programmer: data structures. A PState is an arbitrary combination of data structures, and every PState you create can have a different combination. With the “subindexing” feature of PStates, nested data structures can efficiently contain hundreds of millions of elements. For example, a “map of maps” is equivalent to a “document database”, and a “map of subindexed sorted maps” is equivalent to a “column-oriented database”. Any combination of data structures and any amount of nesting is valid – e.g. you can have a “map of lists of subindexed maps of lists of subindexed sets”. I cannot emphasize enough how much interacting with indexes as regular data structures instead of magical “data models” liberates backend programming.

The last concept in Rama is “query”. Queries in Rama take advantage of the data structure orientation of PStates with a “path-based” API that allows you to concisely fetch and aggregate data from a single partition. In addition to this, Rama has a feature called “query topologies” which can efficiently do real-time distributed querying and aggregation over an arbitrary collection of PStates. These are the analogue of “predefined queries” in traditional databases, except programmed via the same Java API as used to program ETLs and far more capable.

Individually, none of these concepts are new. I’m sure you’ve seen them all before. You may be tempted to dismiss Rama’s programming model as just a combination of event sourcing and materialized views. But what Rama does is integrate and generalize these concepts to such an extent that you can build entire backends end-to-end without any of the impedance mismatches or complexity that characterize and overwhelm existing systems.

All these concepts are implemented by Rama in a linearly scalable way. So if your application needs more resources, you can add them at the click of a button. Rama also achieves fault-tolerance by replicating all data and implementing automatic failover.

Here’s an example of how these concepts fit together. These are all the depots, ETLs, PStates, and query topologies for the portion of our Mastodon implementation handling profiles, statuses, and timelines:

This looks intimidating, but this part of the codebase only totals 1,100 lines of code. And it implements a ton of functionality, all scalably: statuses, timelines, boosts, conversations, favorites, bookmarks, mutes, account registration, profile edits, federation, and more. Notice that the PStates are a diverse collection of data structure combinations, and there are 33 of them here. Many of the ETLs produce multiple PStates and consume multiple depots. Making depots, ETLs, and PStates is inexpensive in Rama and can be done liberally.

What you’re seeing in this diagram is a total inversion of control compared to how applications are typically architected today. For example, consider the “fanout” ETL in this diagram which processes incoming statuses and sends them to follower timelines (you’ll see the code for this later in this post). There are a bunch of rules dictating which statuses go to which followers – boosts never go back to the original author, replies only go to followers who also follow the account being replied to, and so on. Traditionally, that’s accomplished with a “database layer” handling storage and a separate “application layer” implementing the product logic. The “application layer” does reads and writes to the “database layer”, and the two layers are deployed, scaled, and managed separately. But with Rama, the product logic exists inside the system doing the indexing. Computation and storage are colocated. Rama does everything a database does, but it also does so much more.

When building a backend with Rama, you begin with all the use cases you need to support. For example: fetch the number of followers of a user, fetch a page of a timeline, fetch ten follow suggestions, and so on. Then you determine what PState layouts (data structures) you need to support those queries. One PState could support ten of your queries, and another PState may support just one query.

Next you determine what your source data is, and then you make depots to receive that data. Source data usually corresponds to events happening in your application, like “Alice follows Bob”, “James posted the status ‘Hello world’”, or “Bob unfollows Charlie”. You can represent your data however you want, whether Java objects, frameworks like Thrift or Protocol Buffers, or even unstructured formats like JSON (however, we recommend using structured formats as much as possible).

The last step is writing the ETL topologies that convert source data from your depots into your PStates. When deployed, the ETLs run continuously keeping your PStates up to date. Rama’s ETL API, though just Java, is like a “distributed programming language” with the computational capabilities of any Turing-complete language along with facilities to easily control on which partition computation happens at any given point. You’ll see many examples of this API later in this post.

Rama is deployed onto a cluster of nodes. There’s a central daemon called the “Conductor” which coordinates deploys, updates, and scaling operations. Every node has a “Supervisor” daemon which manages the launch/teardown of user code.

Applications are deployed onto a Rama cluster as “modules”. A “module” contains an arbitrary combination of depots, ETLs, PStates, and query topologies. Unlike traditional architectures, where the corresponding analogues exist in separate processes and usually on separate nodes, in Rama these are colocated in the same set of processes. This colocation enables fantastic efficiency which has never been possible before. Modules can also consume data from depots and PStates in other modules just as easily as they can from their own. A module runs forever, continuously processing new data from depots, unless you choose to destroy it.

A module is deployed by giving the Conductor a .jar file with user code. Additionally, configuration is provided for the number of nodes to allocate to the module, replication parameters, as well as any other tuning parameters. A module is updated a similar way: a new .jar is provided with the new code, and the Conductor orchestrates an update sequence that launches new processes and transfers depots and PStates to the new module version.

Here’s what a Rama cluster could look like with two modules deployed, “SocialGraphModule” and “TimelineModule”:

Let’s look at some of the key parts of how Mastodon is implemented on top of Rama. In this section we’ll focus on the design of Mastodon – what PStates are created, the flow of how data is processed, and how everything is organized. You’ll see how Rama’s capabilities enable some seriously novel ways to architect applications. In the next section we’ll look at some of the code from our implementation.

Let’s start with an extremely simple part of the implementation, tracking followers for hashtags. The implementation for this totals 11 lines of code and supports the following queries:

• Does user A follow hashtag H?
• Who follows hashtag H (paginated)?
• How many followers does hashtag H have?

A good way to visualize how data is processed to produce this PState is via a dataflow graph, as you can see how data moves from the start of processing (one or more depots) to the end results of processing (updates to PStates):

The logic here is trivial, which is why the implementation is only 11 lines of code. You don’t need to worry about things like setting up a database, establishing database connections, handling serialization/deserialization on each database read/write, writing deploys just to handle this one task, or any of the other tasks that pile up when building backend systems. Because Rama is so integrated and so comprehensive, a trivial feature like this has a correspondingly trivial implementation.

You may be wondering why the follow and unfollow events go onto the same depot instead of separate depots. Though you could implement it that way, it would be a mistake to do so. Data is processed off a depot partition in the order in which it was received, but there are no ordering guarantees across different depots. So if a user was spamming the “Follow” and “Unfollow” buttons on the UI and those events were appended to different depots, it’s possible a later unfollow could be processed before a prior follow. This would result in the PState ending up in the incorrect state according to the order by which the user made actions. By putting both events in the same depot, the data is processed in the same order in which it was created.

As a general rule, Rama guarantees local ordering. Data sent between two points are processed in the order in which they were sent. This is true for processing data off a depot, and it’s also true for intra-ETL processing when your processing jumps around to different partitions as part of computation.

This dataflow diagram is literally how you program with Rama, by specifying dataflow graphs in a pure Java API. As you’ll see below, the details of specifying computations like this involve variables, functions, filters, loops, branching, and merging. It also includes fine-grained control over which partitions computation is executing at any given point.

Let’s now look at a slightly more involved part of the implementation, the social graph. The social graph totals 105 lines of code and supports the following queries which power various aspects of Mastodon:

• Does user A follow user B?
• How many followers does user A have?
• How many accounts does user A follow?
• Who are user A’s followers in the order in which they followed (paginated)?
• Who does user A follow in the order in which they followed them (paginated)?
• Who is currently requesting to follow user A in the order in which they requested (paginated)?
• Does user A block user B?
• Does user A mute user B?
• Does user A wish to see boosts from user B?

Even though the implementation is so simple with Rama, it’s worth noting that Twitter had to write a custom database from scratch to build their scalable social graph.

Here you can see that account “1” is followed by accounts “2”, “7”, and “5”, with “2” and “7” having boosts enabled and “5” having boosts disabled. Account “4” is only followed by account “1”, and “1” has boosts enabled for that relationship.

The social graph is constructed based on follow, unfollow, block, unblock, mute, and unmute events. In Mastodon’s design, blocking a user also unfollows in both directions (if currently followed). So the block and unblock events go on the same depot as the follow-related events, while the mute and unmute events go on a separate depot.

Here’s the dataflow graph showing how these PStates are computed based on the source data:

This is more involved than the hashtag follows dataflow graph since this ETL supports so many more features. Yet it’s still only 105 lines of code to implement. You can see how this ETL makes use of conditionals, branching, and merging in its implementation. Here’s a few notes on the logic within this ETL:

This ETL interacts with many PStates at the same time. Because of Rama’s integrated nature, these PStates are colocated with one another within the same processes that are executing the ETL logic. So whereas you always have to do a network operation to access most databases, PState operations are local, in-process operations with Rama ETLs. As you’ll see later, you utilize the network in an ETL via “partitioner” operations to get to the right partition of a module, but once you’re on a partition you can perform as many PState operations to as many colocated PStates as you need. This is not only extremely efficient but also liberating due to the total removal of impedance mismatches that characterizes interacting with databases.

Next let’s look at the core of Mastodon, computing and rendering home timelines. This powers the primary page of the Mastodon experience:

This use case is a great example of how to think about building data-intensive systems not just with Rama, but in general. For any backend feature you want to implement, you have to balance what gets precomputed versus what gets computed on the fly at query-time. The more you can precompute, the less work you’ll have to do at query-time and the lower latencies your users will experience. This basic structure looks like this:

This of course is the programming model of Rama you’ve already seen, and a big part of designing Rama applications is determining what computation goes in the ETL portion versus what goes in the query portion. Because both the ETL and query portions can be arbitrary distributed computations, and since PStates can be any structure you want, you have total flexibility when it comes to choosing what gets precomputed versus what gets computed on the fly.

It’s generally a good idea to work backwards from the needed queries to learn how to best structure things. In the case of querying for a page of a timeline, you need the following information:

• Content for each status
• Stats for each status (number of replies, boosts, and favorites)
• Information about the account that posted each status (username, display name, profile pic)
• Whether the author of each status is blocked or muted
• For boosts, the username of the booster
• ID to use to fetch the next page of statuses

Since statuses can be edited and profile information can be changed, almost all this information is dynamic and must be fetched for each render of a timeline page.

Let’s consider a typical way to go about this with non-Rama technologies, even scalable ones, which unfortunately is extremely inefficient:

• Fetch the list of status IDs for a page of the timeline
• In parallel, send database requests to fetch:
  • Content for each status
  • Stats for each status
  • Information for each author

For a page of 20 statuses, this could easily require over 100 database calls with a lot of overhead in the sheer amount of back and forth communication needed to fetch data from each database partition.

We handled this use case with Rama by making use of Rama’s facilities for colocation of PStates. The module is organized such that all information for a status and the account who posted it are colocated in the same process. So instead of needing separate requests for status content, status stats, and author information, only one request is needed per status.

Here are all the depots, PStates, and query topologies in the module implementing timelines and profiles (repeated from above):

This PState is partitioned by the account ID. When I say “partitioned by”, I mean how the Mastodon implementation chooses on which partition to store data for any particular account. The most common way to partition a PState is to hash a partitioning key and modulo the hash by the number of partitions. This deterministically chooses a partition for any particular partitioning key, while evenly spreading out data across all partitions (this “hash/mod” technique is used by most distributed databases). For this PState, the partitioning key is the same as the key in the map, the account ID (as you’ll see soon, it doesn’t have to be).

Because of this colocation queries can look at the partitions for both PStates at the same time within the same event instead of needing two roundtrips. You can also have multiple threads per worker process, or multiple worker processes per node.

Similar to the social graph PStates, this PState is used to compute: the order in which accounts favorited a status (paginated), whether a particular account already favorited a status, and the number of favorites for a status.

Here’s a visualization of how all the PStates described could exist in a module deployment:

This use case is a great example of the power of Rama’s integrated approach, achieving incredible efficiency and incredible simplicity. Each of these PStates exists in exactly the perfect shape and partitioning for the use cases it serves.

Next, let’s explore how home timelines are stored. Materializing home timelines is by far the most intensive part of the application, since statuses are fanned out to all followers. Since the average fanout on our instance is 403, there are over 400x more writes to home timelines than any of the other PStates involved in processing statuses.

PStates are durable, replicated, high-performance structures. They are easy to reason about and ideal for most use cases. In our initial implementation of Mastodon, we stored home timelines in a PState and it worked fine.

However, writing to the home timelines PState was clearly the overwhelming bottleneck in the application because of the sheer amount of data that had to be written to disk and replicated. We also realized that home timelines are unique in that they are somewhat redundant with other PStates. You can reconstruct a home timeline by looking at the statuses of everyone you follow. This can involve a few hundred PState queries across the cluster, so it isn’t cheap, but it’s also not terribly expensive.

As it turns out, what we ended up doing to optimize home timelines is very similar to how Twitter did it for their chronological timelines, and for the exact same reasons. In our revised implementation, we store home timelines in-memory and unreplicated. So instead of needing to persist each timeline write to disk and replicate it to followers, a timeline write is an extremely cheap write to an in-memory buffer. This optimization increased the number of statuses we could process per second by 15x.

Solely storing the timeline in-memory is not sufficient though, as it provides no fault-tolerance in the event of a power loss or other node failure. So a new leader for the partition would not have the timeline info since it’s unreplicated and not persisted to disk. To solve this, we reconstruct the timeline on read if it’s missing or incomplete by querying the recent statuses of all follows. This provides the same fault-tolerance as replication, but in a different way.

Implementing fault-tolerance this way is a tradeoff. For the benefit of massively reduced cost on timeline write, sometimes reads will be much more expensive due to the cost of reconstructing lost timelines. This tradeoff is overwhelmingly worth it because timeline writes are way, way more frequent than timeline reads and lost partitions are rare.

Whereas Twitter stores home timelines in a dedicated in-memory database, in Rama they’re stored in-memory in the same processes executing the ETL for timeline fanout. So instead of having to do network operations, serialization, and deserialization, the reads and writes to home timelines in our implementation are literally just in-memory operations on a hash map. This is dramatically simpler and more efficient than operating a separate in-memory database. The timelines themselves are stored like this:

To minimize memory usage and GC pressure, we use a ring buffer and Java primitives to represent each home timeline. The buffer contains pairs of author ID and status ID. The author ID is stored along with the status ID since it is static information that will never change, and materializing it means that information doesn’t need to be looked up at query time. The home timeline stores the most recent 600 statuses, so the buffer size is 1,200 to accommodate each author ID and status ID pair. The size is fixed since storing full timelines would require a prohibitive amount of memory (the number of statuses times the average number of followers).

Each user utilizes about 10kb of memory to represent their home timeline. For a Twitter-scale deployment of 500M users, that requires about 4.7TB of memory total around the cluster, which is easily achievable.

The in-memory home timelines and other PStates are put together to render a page of a timeline with the following logic:

So that’s the story on how timelines are rendered, but how are home timelines and these various PStates computed? If you look back at the performance numbers, you can see our Mastodon implementation has high throughput that scales linearly while also having great, consistent latencies for delivering statuses to follower timelines.

Let’s focus on how statuses are handled, particularly how timelines are materialized. Whenever someone posts a status, that status must be fanned out to all followers and appended to their home timelines.

The tricky part is dealing with bursty load arising from how unbalanced the social graph can get. In our Mastodon instance, for example, the average fanout is 403, but the most popular user has over 22M followers. 3,500 statuses are posted each second, meaning that every second the system usually needs to perform 1.4M timeline writes to keep up. But if a user with 20M followers posts a status, then the number of timeline writes blows up by 15x to about 21.4M. With a naive implementation this can significantly delay other statuses from reaching follower timelines. And since the latency from posting a status to it reaching follower timelines is one of the most important metrics for this product, that’s really bad!

This is essentially a problem of fairness. You don’t want very popular users to hog all the resources on the system whenever they post a status. The key to solving this issue is to limit the amount of resources a status from a popular user can use before allocating those resources to other statuses. The approach we take in our implementation is:

• For each iteration of timeline processing, fan out each status to at most 64k followers.
• If there are more followers left to deliver to for a status, add that status to a PState to continue fanout in the next iteration of processing.

With this approach a status from a user with 20M followers will take 312 iterations of processing to complete fanout (about 3 minutes), and fairness is achieved by giving all statuses equal access to resources at any given time. Since the vast majority of users have less than 64k followers, most users will see their statuses delivered in one iteration of processing. Statuses from popular users take longer to deliver to all followers, but that is the tradeoff that has to be made given that the amount of resources is fixed at any given time.

As a side note, Twitter also has special handling for users with lots of followers to address the exact same issue.

With the basic approach understood, let’s look specifically at how statuses are processed in our implementation to materialize timelines. Here’s a dataflow diagram showing the logic:

This dataflow diagram is a little bit different than the social graph one, as this ETL is implemented with microbatching while the social graph is implemented with streaming. Streaming processes data directly off of depots as it arrives, while microbatching processes data in small batches. “Start iteration” in this diagram signifies the beginning of a microbatch. The tradeoffs between streaming and microbatching are too much to cover for this post, but you’ll be able to learn more about that next week when we release the documentation for Rama.

In this diagram you can see all the steps involved in delivering statuses to followers. There are many product rules implemented here, such as: only fan out statuses with the correct visibility, only send replies to followers who also follow the account being replied to, respect “show boost” settings, and so on.

This dataflow diagram only shows the computation of home timelines, whereas in reality this ETL also handles lists, hashtag timelines, and conversations. Those all work similarly to home timelines and are just additional branches of dataflow computation with slightly differing logic.

Fairness issues aren’t the only problem caused by an unbalanced social graph. Another problem is skew: a naive implementation that handles all fanout for a single user from a single partition will lead to some partitions of a module having a lot more overall work to do than others. Without balanced processing, throughput is lowered since some resources are idle while others are being overwhelmed.

In our Mastodon implementation, we put significant effort into balancing fanout computation regardless of how unbalanced a social graph gets. This is a deep topic, so we will explore this further in a subsequent blog post. The optimizations we did in this area increased throughput by about 15%.

Let’s now look at another part of Mastodon which works completely differently than timelines: personalized follow suggestions. This powers this section on Mastodon:

Everything about how data is processed and indexed for follow suggestions is different from what we looked at in the previous sections. Taken together, the implementations of timelines, the social graph, and follow suggestions demonstrate how expressive Rama is as a system. This generality is a result of the total arbitrariness with which you can write ETL computations, structure indexes, and compute queries.

Unlike the social graph and timelines, the behavior of personalized follow suggestions could be specified in very different ways. We’ve chosen to determine follow suggestions like so:

• Rank accounts to suggest based on who’s most followed by the accounts you already follow
• Don’t suggest accounts you already follow
• If this doesn’t produce enough suggestions (e.g. because you don’t follow many accounts yet), suggest the most followed accounts on the platform

We took a different approach than Mastodon for follow suggestions – the API docs describe follow suggestions as “Accounts that are promoted by staff, or that the user has had past positive interactions with, but is not yet following.” We chose our approach because it’s much more difficult to implement and thus a better demonstration of Rama. Our implementation of personalized follow suggestions totals 141 lines of code.

Follow suggestions can’t be computed fully incrementally, at least not practically. That is, you can’t receive a new follow or unfollow event and incrementally update all the follow suggestions for affected accounts. Computing follow suggestions is a batch operation that needs to look at everyone an account follows, and everyone they follow, at the same time in a single computation.

With that in mind, there are a few pieces to our implementation. First, everyone’s follow suggestions are recomputed on a regular basis. The ETL for follow suggestions recomputes the suggestions for 1,280 accounts every 30 seconds. Since there are 100M accounts, this means each account has its suggestions updated every 27 days.

In addition to this, we have special handling for new users. When a new user signs up, you want to provide good follow suggestions as soon as possible in order to increase engagement and increase the chance they’ll continue to use the service. So you don’t want to wait 27 days to compute personalized suggestions for a new user. At the same time, you can’t produce good personalized suggestions until the user has followed at least a few accounts. So our implementation tracks milestones which trigger immediate recomputation of follow suggestions: when a user follows 10 accounts and when a user follows 100 accounts.

Here are all the depots, PStates, and query topologies related to the follow suggestions implementation:

Here’s what each PState for follow suggestions is used for:

Notice how some of these PStates are not maps at the top-level, which may feel unusual given that pretty much every database that’s ever existed is map-based (with a “key” being the central concept to identify a record or row). But just as data structures other than maps are useful for everyday programming, data structures other than maps are useful for backend programming as well.

You can think of time as an infinite stream of events, with each event being an instant in time. Rama exposes a special depot for time called a “tick depot” which emits events according to a specified frequency.

For follow suggestions, a tick depot is used to trigger processing every 30 seconds. The subsequent computation then uses the PStates described to recompute follow suggestions for a subset of users. The dataflow looks like this:

In between recomputes, a user may have followed some of the users in their list of suggestions. This is handled with a query topology that filters a user’s follow suggestions to exclude users they already follow.

This submits the module and its code to the cluster with the given parallelism, and Rama then launches worker processes around the cluster to run the module. The same cluster is shared by all modules. Once the workers finish starting up, they start reading from depots, executing ETLs, updating PStates, serving query requests, and so on.

The “replication factor” specifies to how many nodes each depot and PState should replicate its data. Replication happens completely behind the scenes and provides automatic failover in case of failures (e.g. hardware issues). Rama provides very strong guarantees with replication – data is not made visible for consumption from depots or PStates until it has been successfully replicated.

After the module finishes launching, the Cluster UI for Rama starts displaying telemetry on the module:

Rama tracks and displays telemetry for the module as a whole, as well as specific telemetry for each topology, depot, and PState. The telemetry is extremely useful for understanding the performance of a module and when it needs to be scaled. Rama uses itself to implement telemetry – a built-in module collects telemetry data from all modules into a depot, processes that data with an ETL, and indexes the results into PStates arranged in a time-series structure.

When we want to update the module to add a feature (e.g. add a new PState) or fix a bug, we run a command like the following:

This launches a carefully coordinated automated procedure to launch new worker processes and handoff responsibility for depots and PStates to the new version of the module. Clients of the module doing depot appends and PState queries don’t need to be updated and automatically transition themselves to the new module version.

Similarly, when we want to scale the module to have more resources, we run a command like the following:

This launches a similar procedure as module update to transition the module to the new version.

And that’s all there is to DevOps with Rama – it’s just a few commands at the terminal to manage everything. You don’t need to invest huge amounts of time writing piles of shell scripts to coordinate changes across dozens of systems. Since Rama is such a cohesive, integrated system it’s able to automate deployment entirely, and it’s able to provide deep and detailed runtime telemetry without needing to lift a finger.

Let’s look at some code! Before diving into the Mastodon code, let’s look at a simple example of coding with Rama to gain a feeling for what it’s like.

I’m not going to explain every last detail in this code – the API is so rich that it’s too much to explain for this post. Instead, I’ll do my best to summarize what the code is doing. In one week we will be releasing all the documentation for Rama, and this includes a six part tutorial that gently introduces everything. We will also be releasing a build of Rama that anyone can download and use. This build will be able to simulate Rama clusters within a single process but will not be able to run distributed clusters. It has the full Rama API and can be used to experiment with Rama. Once we open-source our Mastodon implementation in two weeks, you’ll be able to run it within a single process using this build.

With that said, let’s look at a simple example. Here’s the complete definition for a “word count module”, which accepts sentences as input and produces a single PState containing the count of all words in those sentences:

Finally, now that the computation is on the correct partition, the last line updates the count for the word in the PState. This PState update is specified in the form of an aggregation template – in this case it says it’s aggregating a map where the key is the word and the value is the count of all events seen for that word.

This is such a basic example that it doesn’t really do justice to the expressive power of Rama. However, it does demonstrate the general workflow of declaring modules, depots, PStates, and topologies. Some of the functionality not shown here includes: consuming depots/PStates from other modules, query topologies, microbatching, branching/merging, joins, loops, shadowing variables, conditionals, and decomposing code with macros.

Let’s now take a look at interacting with Rama modules as a client outside the cluster, similar to how you interact with a database using a database client. Here’s code that connects to a remote cluster, creates handles to the depot and PState of the module, appends some sentences, and then does some PState queries:

The PState queries here fetch the values for the specified keys. PStates are queried using Rama’s “Path” API, and this example barely scratches the surface of what you can do with paths. They allow you to easily reach into a PState, regardless of its structure, and retrieve precisely what you need – whether one value, multiple values, or an aggregation of values. They can also be used for updating PStates within topologies. Mastering paths is one of the keys to mastering Rama development.

Running this code prints:

Now let’s look at some code from our Mastodon implementation. We’ll be looking at bigger code samples in this section utilizing a lot more of the Rama API, so even more than the last section I won’t be able to explain all the details of the code. I’ll summarize what the key parts are, and you’ll be able to learn all the details next week when we release the documentation.

Please don’t be too intimidated by this code. There are a lot of concepts and API methods at work here, and no one could possibly understand this code completely at a first glance. This is especially true without the accompanying documentation. I’m showing this code because it ties together the high-level concepts I’ve discussed in this post by making them real instead of abstract.

Let’s start by looking at an example of how data is defined. We chose to represent data using Thrift since it has a nice schema definition language and produces efficient serialization, but you can just as easily use plain Java objects, Protocol Buffers, or anything else you want. We use Thrift-defined objects in both depots and PStates. Here’s how statuses are defined:

Every type of status, including boosts, replies, and statuses with polls is represented by this definition. Being able to represent your data using normal programming practices, as opposed to restrictive database environments where you can’t have nested definitions like this, goes a long way in avoiding impedance mismatches and keeping code clean and comprehensible.

Next, let’s look at the entire definition of following hashtags (described earlier in this post). As a reminder, here’s the dataflow diagram for the following hashtags ETL:

Here’s the implementation, which is a direct translation of the diagram to code:

Unlike word count, this code uses paths instead of aggregators to define the writes to the PStates, which is the same API used to read from PStates. You’ll be able to learn more next week when we release Rama’s documentation about the differences between aggregators and paths and when to prefer one over the other.

Note that this code defines a parallel computation just like the word count example earlier. The code runs across many nodes to process data off each partition of the depot and update the PState. Any failures (e.g. a node dying) are handled transparently and Rama guarantees all depot data will be fully processed.

Next, let’s look at the entire definition of the social graph as described earlier. Here was the dataflow diagram for the social graph ETL:

Like hashtag follows, the social graph implementation is also a direct translation of the diagram to code. Since the code for this is longer, let’s look at it section by section in the order in which it’s written. The first part is the declaration of the depots, topology, and PStates:

Note that both hashtag follows and the social graph are part of the same stream topology. The social graph implementation consumes different depots than hashtag follows does, so the code is otherwise completely independent.

The next part defines the root of processing where the branching occurs at the start of the dataflow diagram:

As we build up the code for the social graph, let’s also take a look visually at how the dataflow diagram is filled out. This code starts off the dataflow diagram like this:

This branch passes all data through to the anchor “Normal”, which will later be merged with other branches as you can see in the dataflow diagram.

The next section defines the branch handling implicit unfollow events for block events:

The next section defines the branch handling accepting a follow request:

The next section handles follows to a locked account. As a reminder, a locked account requires all followers to be manually approved.

If the follow relationship does not already exist, then the PState tracking follow requests is updated.

The next section merges the prior branches together and begins processing for the rest of the events, whether they came directly off the depot or were created implicitly by one of the branches:

This code just removes the relationship from all relevant PStates.

Here is the next section:

This just removes the follow request from the PState.

That’s the complete implementation of the social graph! As you can see, it’s expressed exactly as you saw in the dataflow diagram with branches, merges, and dispatching on the types of events.

The code for this ETL is also a great example of why it’s so beneficial to interact with your data layer with an API in a general-purpose language instead of a custom language (like SQL). This code makes use of normal programming practices to factor out reusable functionality or to separate code into separate functions to make it easier to read. This code also demonstrates how easy it is to intermix logic written in Java with logic written in Rama’s dataflow API. Method references, lambdas, and macros are facilities for combining the two.

Let’s look at some of the Mastodon API implementation related to the social graph so you can see how you interact with a Rama cluster to serve the frontend. Here’s how a new unfollow event is added:

Here’s how a follow event is handled, conditioning the type of data appended depending on if the account is locked or not: