Next-level backends with Rama: fault-tolerant timed notifications in 25 LOC

Backend storage

This uses a higher-level abstraction called TopologyScheduler from the open-source rama-helpers project. TopologyScheduler is a small helper that handles the storage and logic for scheduling future work, designed to be used as part of any module. It keeps state in a PState which is declared into a topology with the declarePStates method. You’ll soon see its helper methods for injecting code into a topology to schedule future work and handling scheduled work which is ready to execute. Because TopologyScheduler is built upon Rama’s primitives of PStates and topologies, it’s scalable and fault-tolerant.

declarePStates takes as an argument the name of the PState for this TopologyScheduler instance. Later on when methods are called on TopologyScheduler to inject code, it will automatically reference that PState when doing reads and writes. TopologyScheduler stores in that PState an ordered queue based on the timestamp for which each piece of future work is scheduled.

Let’s now review the broader concepts of Rama in order to understand how these PStates will be utilized.

Rama concepts

Implementing the module

This declares a Rama module called “TimedNotificationsModule” with two depots. The first depot is called *scheduled-post-depot and will receive all scheduled post information. Objects appended to a depot can be any type. The second argument of declaring the depot is called the “depot partitioner” – more on that later.

To keep the example simple, the data appended to the depot will be defrecord objects for the Clojure version and HashMap objects for the Java version. To have a tighter schema on depot records you could instead use Thrift, Protocol Buffers, or a language-native tool for defining the types. Here’s the function that will be used to create depot data:

The second depot *tick will be used for checking if any scheduled posts are ready to be appended to a user’s feed and then triggering the appropriate code to do so. Unlike the first depot, this is declared as a “tick depot”. Whereas a normal depot emits whenever new data is appended to it, a tick depot emits when the configured amount of time has passed. Subscribing to a tick depot from a topology is no different than subscribing to a regular depot. This particular tick depot is configured to emit once every 1000 milliseconds, meaning a scheduled post will be delivered within one second of its scheduled time.

Next, let’s begin defining the topology to consume data from the depot and materialize the PStates. Here’s the declaration of the topology with the PStates:

Let’s now add the code that consumes *scheduled-post-depot to get posts durably scheduled for the future:

This subscribes the topology to the depot *scheduled-post-depot and starts a reactive computation on it. Operations in dataflow do not return values. Instead, they emit values that are bound to new variables. In the Clojure API, the input and outputs to an operation are separated by the :> keyword. In the Java API, output variables are bound with the .out method.

Whenever data is appended to that depot, the data is emitted into the topology. The Java versions binds the emit into the variable *scheduled-post and then gets the field “time-millis” from the map into the variable *time-millis , while the Clojure version captures the emit as the variable *scheduled-post and also destructures a field into the variable *time-millis . All variables in Rama code begin with a * . The subsequent code runs for every single emit.

Remember that last argument to the depot declaration called the “depot partitioner”? That’s relevant here. Here’s that image of the physical layout of a module again:

The depot partitioner determines on which task the append happens and thereby on which task computation begins for subscribed topologies. In this case, the depot partitioner says to hash by the “id” field of the appended data. The target task is computed by taking the hash and modding it by the total number of tasks. This means data with the same ID always go to the same task, while different IDs are evenly spread across all tasks.

Rama gives a ton of control over how computation and storage are partitioned, and in this case we’re partitioning by the hash of the user ID since that’s how we ultimately want the $$feeds PState to be partitioned. This allows us to easily locate the task storing data for any particular user.

The final line schedules the post for the future:

This uses the scheduleItem method from TopologyScheduler to write the scheduled post into the PState managed by TopologyScheduler . scheduleItem returns a block of code that is inserted into the topology using Rama’s “macro” facility. “Macros” are a feature from Rama’s Java API for decomposing bits of dataflow code and later mixing them in to any other dataflow code. The definition of scheduleItem from the TopologyScheduler implementation is as follows:

Here you can see it writes to its managed PState using localTransform after doing a bit of preparatory logic. Any intermediate variables are given unique names using genVar so they don’t inadvertently shadow any variables that are already in scope in the dataflow code where this code is being inserted. You can read more about macros in this section. Note that the Clojure API has additional facilities for decomposition that the Java API does not have, such as deframaop.

Understanding the implementation of this macro isn’t important – what matters is it handles the work of scheduling an item for the future at a particular timestamp. scheduleItem takes in a timestamp and an item, and that item will be given to the callback that’s later invoked when the scheduled time is reached. The state for scheduling the item is stored on the PState on this task, and the callback will be invoked on this same task when it’s ready.

Since TopologyScheduler is built with the Java API, the Clojure API has the java-macro! operation so that it can use utilities built around the Java API. This is also why the variables are specified as strings instead of symbols in this line of Clojure code, since the Java API represents variables as strings beginning with * .

Let’s now take a look at handling callbacks from TopologyScheduler , which completes the module implementation:

This first adds a subscription to the tick depot, which emits at the configured 1000 millisecond frequency. The emit isn’t captured into a variable like the previous depot subscription since all this code cares about is the frequency at which the code runs.

Tick depots are global and emit only on task 0, so the subsequent code runs on just one task each time it emits. The next line handles expired items using the handleExpirations method from TopologyScheduler and inserting the code into the topology with a macro. handleExpirations does the following:

handleExpirations takes in three arguments as input. The first specifies a variable to bind the expired item. This is the same item as was passed to scheduleItem before, and the variable will be in scope for the callback code. The next argument specifies a variable to bind the current time, which isn’t needed for this particular use case. The last argument is the callback code that specifies what to do with an expired item. It’s an arbitrary block of dataflow code. Since the method expects a Java API block of code, the Clojure API uses java-block<- to convert Clojure dataflow code to a Java API block.

Let’s go through each line of this callback code:

First, the “id” and “post” fields from the scheduled post are extracted into the variables *id and *post . The Clojure API destructures them, while the Java API uses the Ops.GET function to fetch them from the map.

The final line adds the post to the feed for the user:

The PState is updated with the “local transform” operation. The transform takes in as input the PState $$feeds and a “path” specifying what to change about the PState. When a PState is referenced in dataflow code, it always references the partition of the PState that’s located on the task on which the event is currently running.

Paths are a deep topic, and the full documentation for them can be found here. A path is a sequence of “navigators” that specify how to hop through a data structure to target values of interest. A path can target any number of values, and they’re used for both transforms and queries. In this case, the path navigates by the key *id to the list of posts for that user. The next navigator, called AFTER-ELEM in Clojure and afterElem() in Java, navigates to the “void” element after the end of the list. Setting that “void” element to a value with the “term val” navigator causes that value to be appended to that list.

This code writes to the correct partition of $$feeds because the callback code is run on the same task where the post was scheduled. That was on the correct task because of the depot partitioner, as discussed before.

TopologyScheduler performance

TopologyScheduler adds very little overhead to processing. Scheduling an item of future work is a fast PState write, and checking for expired items is also very fast due to how it’s indexed by TopologyScheduler . The total overhead from TopologyScheduler is one small unit of work to schedule it, and then one small unit of work to emit it to the callback code later. This means TopologyScheduler can handle very large throughputs of scheduled items.

Summary