Next-level backends with Rama: personalized content moderation in 60 LOC

Backend storage

This declares the PState as a map of lists, with the keys being 64-bit user IDs and the list values being post data. The name of a PState always begins with $$ . For each user, it tracks their feed as a list of content. The inner list is declared as “subindexed”, which instructs Rama to store the elements individually on disk rather than the whole list read and written as one value. Subindexing enables nested data structures to have billions of elements and still be read and written to extremely quickly. This PState can support many queries in less than one millisecond: get the number of posts on a user’s feed, get a single post at a particular index, or get all posts between two indices.

The second PState tracks all the users muted by each user:

Rama concepts

Materializing the PStates

This declares a Rama module called “ContentModerationModule” with two depots. The first depot is called *post-depot and will receive all “post” data. 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.

The second depot is called *mute-depot and will receive mute and unmute events.

Part of designing Rama modules is determining how many depots to have and how to split different types of data across depots. In this case, post objects go into one depot while both mute and unmute events go in the same depot. Data should go into the same depot when there’s an ordering relationship between them, and they should otherwise go into separate depots so consumers don’t need to filter out data they’re not interested in. There’s an ordering relationship between mutes and unmutes – imagine a user were spamming the mute/unmute buttons over and over causing tons of events to be appended. If they were sent to separate depots, they could be processed in a different order than they were appended, producing the wrong result. Putting them in the same depot ensures they’re processed in the exact order they were appended. On the other hand, there’s no ordering relationship between posts and mutes/unmutes, so they can go in separate depots.

To keep the example simple, the data appended to the depots 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 functions that will be used to create depot data:

Let’s now add the code to materialize the $$posts PState:

This subscribes the topology to the depot *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 *post and then gets the fields “to-user-id” from the map into the variable *to-user-id , while the Clojure version captures the emit as the variable *post and also destructures the “to-user-id” field into the variable *to-user-id . 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 PState is updated with the “local transform” operation. The transform takes in as input the PState $$posts 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 *to-user-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.

Let’s now add the code to materialize the $$mutes PState:

This adds a new depot subscription to *mute-depot to the same topology. Topologies can subscribe to any number of depots, and each source subscription is processed independently. This code binds emits from that depot to the variable *data .

This depot emits “mute” and “unmute” events. Mutes should add to the PState, while unmutes should remove from the PState. The subsequent code adds a conditional on the type of *data :

The Java and Clojure versions implement this a little differently since the Clojure version is using first-class types while the Java version is using plain maps with a “type” key inside.

The Clojure version uses <<subsource , which is a convenient way to dispatch code on the type of an object. You could also use <<if or <<cond for this, but the <<subsource version is a little more concise.

The Java version fetches the “type” and “user-id” fields from the map into the variables *type and *user-id . *user-id will be used a little later. It then uses .ifTrue to check if *type is equal to “mute”. The “then” branch of that conditional will handle mutes, and then “else” branch will handle unmutes.

The next bit of code handles mutes:

In the Clojure version, case> begins a handler for a particular type given to the surrounding <<subsource form. This case> handles Mute objects. case> allows the object given to <<subsource to be further destructured in its emit. Here it destructures *user-id and *muted-user-id from *data .

The Java version fetches the “muted-user-id” field from the map into the variable *muted-user-id .

Then, the “local transform” operation is invoked to update the $$mutes PState. The path used is similar to the path from the earlier “local transform”, but the inner data structure being updated is a set instead of a list. The path first navigates to the inner set by the key *user-id . The next navigator, called NONE-ELEM in Clojure and voidSetElem in Java, navigates to the “void” element of the set. Setting that “void” element to a value causes that value to be added to that set.

Adding to a set is idempotent, so if the frontend were to append multiple mutes for the same user in a row, the mute for that user will only exist one time in the set.

The last part of the topology handles unmutes:

This is the inverse of the previous case. The Clojure version declares a case> to handle Unmute objects and destructure *user-id and *unmuted-user-id . The Java version is the “else” branch of the .ifTrue and starts by fetching “unmuted-user-id” from the map into the variable *unmuted-user-id .

Then, a “local transform” is done to remove the user from the set of mutes. The path first navigates to the inner set by the key *user-id . The next navigator, called set-elem in Clojure and setElem in Java, navigates to the specified element of the set if it exists. If the element doesn’t exist, the path navigates nowhere and the transform is a no-op. The last navigator removes the element from the set. Removing elements with paths is done in Clojure with NONE> and in Java with .termVoid() . This navigator sets the element to “void”, which causes it to be removed from its containing data structure.

That completes the materialization of $$posts and $$mutes . Now we’re ready to move on to implementing an efficient query to fetch a page from $$posts with muted content filtered out.

Implementing the query

1. Set N=10, START_OFFSET=75
2. Read posts for user ID 123 from PState $$posts from indices START_OFFSET to START_OFFSET + N
3. For each post, check if the poster is muted by user ID 123 in the $$mutes PState
4. Set NUM_FETCHED_POSTS equal to the number of posts remaining after filtering
5. If NUM_FETCHED_POSTS is less than N, go back to step 2 with N set to (N – NUM_FETCHED_POSTS) and START_OFFSET set to START_OFFSET + N

This algorithm iteratively fetches content from the $$posts PState until the desired number of posts is found.

In the case of a user having a ton of mutes set as well as a lot of posts, it’s possible this algorithm could take a long time to scan through the $$posts PState to find the desired number of posts. If this is a concern, the algorithm could be enhanced to have an upper limit on the number of times it loops. This implementation won’t do that, but it’s easy to add if needed.

This will be implemented via a predefined query in the module called a “query topology”. Query topologies are programmed with the exact same dataflow API as we already used to implement the ETL topology. All the behavior of this algorithm – looping, checking posts to see if they should be filtered, and aggregating results will all be done within the module. This means the client doing the query only has to make one request to the backend. If the client implemented this algorithm client-side, it would have to make many requests adding a lot of overhead for all the roundtrips. Implementing as a query topology eliminates all that overhead which minimizes the latency of the query.

Query topologies are like functions in Java or Clojure. They take in any number of input arguments and return one value as the result. As such, query topologies can be decomposed into multiple query topologies just like you can decompose a function’s implementation into multiple functions. This query will be implemented with two query topologies. The first will fetch a page of posts and filter them according to mute settings. The second will use the first query topology and loop until the desired number of posts have been found.

Implementing the first query topology

This declares a query topology named “get-posts-helper” that takes in input arguments *user-id , *start-offset , and *end-offset . It declares the return variable *posts , which will be bound by the end of the topology execution.

The next line gets the query to the task of the module containing the data for that user ID:

The line does a “hash partition” by the value of *user-id . Partitioners relocate subsequent code to potentially a new task, and a hash partitioner works exactly like the aforementioned depot partitioner. The details of relocating computation, like serializing and deserializing any variables referenced after the partitioner, are handled automatically. The code is linear without any callback functions even though partitioners could be jumping around to different tasks on different nodes.

When the first operation of a query topology is a partitioner, query topology clients are optimized to go directly to that task. You’ll see an example of invoking a query topology in the next section.

The next bit of code fetches all posts in the requested range:

A “local select” does a query on the partition of the PState that’s on the same task on which the event is running. This path first navigates to the list of posts for the key *user-id . It then navigates to the “sublist” between *start-offset and *end-offset . At this point the path is navigated to a single list of values. The last navigator navigates to each individual element of that sublist, causing the “local select” to emit one time for each post in that range.

What’s happening here is very different from “regular” programming, where you call a function which returns one value as the result. In dataflow programming, operations can emit any number of times. You can think of the output of an operation as being the input arguments to the rest of the topology. When an operation emits, it invokes the “rest of the topology” with those arguments (the “rest of the topology” is also known as the “continuation”). This is a powerful generalization of the concept of a function, which as you’re about to see enables very elegant code.

The code after the “local select” is now operating on a single post at a time, and you’ll see how the results are aggregated together at the end. The output of the “local select” is bound to the variable *post . The Clojure version destructures the field from-user-id from that object into the variable *from-user-id , and the Java version fetches the field “from-user-id” from the map into the variable *from-user-id .

The next line checks to see if that user is muted:

This code does a “local select” on the $$mutes PState. Since the $$mutes PState stores data with the same partitioning scheme as the $$posts PState – by the hash of the top-level key – the code does not need to switch to a different task with a partitioner before querying.

The query emits a single boolean to the variable *muted? on whether that user who made the post is muted or not. The path first navigates to the inner set by the key *user-id . The next navigator, view , runs a function on the navigated value. The function can be any Java or Clojure function, and it’s given as arguments the navigated value plus any additional arguments given to view . In this case it just checks if the posting user ID is contained in the set.

The next line filters out posts from muted users:

This operation, filter> in Clojure and .keepTrue in Java, emits exactly one time if its input is true and otherwise doesn’t emit. Just like how the “local select” call differs from regular functions by emitting many times, this differs from regular functions by potentially not emitting at all. No values are captured for the output since this operation doesn’t emit any values when it emits, but you can still think of its emit as invoking the “rest of the topology”.

In this case, the code only continues for users that aren’t muted.

The next line is:

This particular aggregator combines all the remaining posts into a single list into the variable *posts . *posts is the same variable name as declared earlier for the output of the query topology.

This completes the first query topology.

Implementing the second query topology

This declares the query topology “get-posts” with input arguments *user-id , *from-offset , and *limit . It declares the return variable *ret , which will be bound by the end of the topology execution.

Just like the first query topology, the next line gets the query to the task of the module containing the data for that user ID:

Loops in dataflow first declare “loop variables” which are in scope for the body of the loop and are set to new values each time the loop is recurred. The Clojure version declares two loop variables, *query-offset and *posts , which are initialized to *from-offset and [] respectively. The Java version will collect posts into a mutable ArrayList , so it initializes that ArrayList into the variable *posts before the loop. The Java version’s only loop var is *query-offset which is initialized to *from-offset .

Loops can be emitted from any number of times (with :> in Clojure and .emitLoop in Java), and each emit runs the code after the loop with the emitted values. The Clojure version binds the emitted values from this loop as *posts and *next-offset in the binding vector. The Java version specifies emitted loop values at the end of the loop body. Again, since the Java version is accumulating posts into a mutable value its loop will only emit the one value *next-offset .

This query topology will return the fetched posts as well as this *next-offset value. In order for the frontend to paginate to the next page of posts, it needs to know where the prior query topology left off in its traversal. Returning *next-offset here lets the frontend know exactly where to begin for the next page. Additionally, this query topology will return “null” for *next-offset if it reached the end of the list of posts, which indicates to the frontend that it reached the last page.

The first query topology needs to know the range of offsets to fetch from the subindexed list of posts. The next line starts to prepare for that by querying for the total number of posts in the PState for this user:

This “local select” call is similar to the earlier one, using view to run a function on the nested list to get the total number of posts and bind that to *num-posts . Fetching the size of a subindexed list is a fast sub-millisecond query even if the list has billions of elements.

The next two lines determine the last offset that will be fetched:

The number of posts remaining to fetch is simply the number of desired posts, *limit , subtracted by the number of posts fetched so far. This is bound to the variable *fetch-amount . The last offset to query is the minimum of the last offset in the subindexed list and *fetch-amount offsets after the query offset. The result of that is bound to *end-offset .

The next line invokes the first query topology to get the posts between those offsets from non-muted users:

The Clojure version appends all the fetched posts into *posts to produce the new variable *new-posts . The Java version uses a lambda to add the fetched posts into the ArrayList with its addAll method. It gives the lambda the arguments from dataflow *posts and *fetched-posts , which become the arguments to the lambda. This code shows how arbitrary Java code can be injected into dataflow, and you can also do so with method references as well.

The code now needs to determine what to do depending on the current state. There are two cases to check: “has it reached the end of the list of posts?”, and “has it found the desired number of posts?”. If the answer to either of those is true, traversal is finished and it should finish the query topology execution. Otherwise, it should continue traversing.

You could implement this with nested “if” conditionals ( <<if in the Clojure API and .ifTrue in the Java API), but Rama provides a more elegant way to express code that needs to check more than one condition through its “cond” operation. The next code begins this form of conditional like so:

It’s reached the end when the *end-offset computed earlier is the same as the total number of posts. The subsequent code is the body of the “case”, which can be any amount of arbitrary dataflow code. In this case, there’s only one line which says to emit from the loop. As explained earlier, the loop emits the “next offset” which will be included in the return value so the frontend knows where to begin in order to query the next page of content. The Clojure version emits the full accumulation of posts along with nil for the “next offset”, which indicates it reached the end of the list of posts. The Java version just emits null for the “next offset”.

The next case checks if the desired number of posts from non-muted users was found:

The desired number of posts has been found if the size of the list of accumulated posts is equal to the *limit argument for the query topology. The body of this case emits from the loop. The Clojure version emits the full accumulation of posts along with *end-offset for the “next offset”. The Java version just emits *end-offset for the “next offset”.

Finally, the last case invokes another iteration of the loop:

Using default> in the Clojure API is the same as saying (case> true) . The body of this case invokes the next iteration of the loop with continue> in the Clojure API and continueLoop in the Java API. This runs the loop from the start with new values for the loop vars.

The .out in the Java version binds the output of the loop into the variable *next-offset , which in the Clojure version was specified in the binding vector at the beginning of the loop.

The next line, which is after the loop, invokes the required “origin partitioner”:

This map is bound to the variable *ret , which is the same as the output variable for the query topology specified in the beginning.

This completes the definition of the query to fetch a page of content for a user. Since it’s executing entirely in the module, the query is extremely efficient since the client only has to make a single request. All the overhead of making network requests and serializing/deserializing values back and forth is completely avoided.

Interacting with the module

Summary