Supplement · The X algorithm · page 2 of 11

The pipeline: how the feed is assembled from standard stages

The «For you» feed is described as a sequence of 75 stages of six types. Let us look at the framework that holds it together: what runs in parallel and what is obliged to go in order, what happens when a source falls over, and why query hydration is split into two rounds.

1. Six types of stage

The whole framework is 1234 lines across seven files. Each file describes one type of stage, and they are all built the same way.

pub trait Filter<Q, C>: Any + Send + Sync {
    fn enable(&self, _query: &Q) -> bool { true }
    fn filter(&self, query: &Q, candidates: Vec<C>) -> FilterResult<C>;
    fn name(&self) -> &'static str { ... }
}

A fragment of filter.rs · code by X, Apache 2.0, commit 28e414f

Three methods: whether it is on, what it does, what it is called. The name is needed for the metrics — it is what the counts of how much this stage dropped and how long it took are attributed to. The run method the pipeline calls is a wrapper around filter that adds the measurements.

2. The order of execution

The main function of the pipeline reads like a table of contents of the whole feed:

let hydrated_query = self.hydrate_query(query).await;
let hydrated_query = self.hydrate_dependent_query(hydrated_query).await;
let candidates = self.fetch_candidates(&hydrated_query).await;
let hydrated_candidates = self.hydrate(&hydrated_query, candidates).await;
let (kept, mut filtered) = self.filter(&hydrated_query, hydrated_candidates.clone());
let scored = self.score(&hydrated_query, kept).await;
let SelectResult { selected, non_selected } = self.select(&hydrated_query, scored);
let post_selection = self.hydrate_post_selection(&hydrated_query, selected).await;
let (mut final_candidates, post_filtered) = self.filter_post_selection(&hydrated_query, post_selection);

A fragment of candidate_pipeline.rs · code by X, Apache 2.0, commit 28e414f

What is parallel and what is not

The difference is visible right in the code and it is not cosmetic.

The sources — in parallel:

let source_futures = sources.iter().map(|s| s.run(query));
let results = join_all(source_futures).await;

The filters — sequentially:

for filter in enabled {
    let result = filter.run(query, candidates);
    ...
}

A fragment of candidate_pipeline.rs · code by X, Apache 2.0, commit 28e414f

Why so? The sources do not depend on each other: each takes the request and returns a list. They can be started simultaneously, and the total time becomes the time of the slowest rather than the sum.

The filters do depend: each receives what is left after the previous one. They could be run in parallel on one and the same input — the result would in theory be the same — but then every filter would process the full list instead of a truncated one. With thirty filters and thousands of candidates that is more expensive, not less. Besides, the order carries meaning: the cheap filters stand earlier and reduce the work for the expensive ones.

3. What happens when a source falls over

A line that is easy to skip past but which determines the behaviour of the system under load:

let results = join_all(source_futures).await;

let mut collected = Vec::new();
for mut candidates in results.into_iter().flatten() {
    collected.append(&mut candidates);
}

A fragment of candidate_pipeline.rs · code by X, Apache 2.0, commit 28e414f

Every source returns Result<Vec<C>, String> — either candidates or an error. And flatten over an iterator of results silently throws out every error and keeps only the successful ones.

So if the service of fresh posts from follows has fallen over, the request does not fall. The feed will be assembled from what the other sources returned: it will get worse — there will be no posts from the people you follow — but it will exist.

4. Query hydration: two rounds

Note that query hydration is called twice:

let hydrated_query = self.hydrate_query(query).await;
let hydrated_query = self.hydrate_dependent_query(hydrated_query).await;

A fragment of candidate_pipeline.rs · code by X, Apache 2.0, commit 28e414f

The first round is the hydrators that need only the original request: the lists of follows, blocks, muted accounts, demographic data. They are all parallel to each other.

The second round is those that need the results of the first. To count mutual follows, for example, you first have to get the list of follows. Or to build the sequence of actions for the model, you have to know which posts have already been shown.

This is the simplest form of dependency planning: instead of a full graph, two levels. The decision is pragmatic: a real dependency graph would require describing the links between stages, while two levels cover almost every real case and require nothing beyond putting a hydrator into the right list.

What exactly is pulled onto the request

There are twenty query hydrators. Here are the most telling ones.

5. Two pipes: posts and everything else

There are in fact two pipelines, and they are nested one inside the other.

The Post Pipeline is what we discussed above: 75 stages that find and order posts.

The Blending Pipeline is a wrapper around it. For it, the ranked posts are just one source among others. The other sources bring what the model does not rank: ads, the «who to follow» block, promo prompts.

6. Side effects: after the answer

fn run_side_effects(&self, input: Arc<SideEffectInput<Q, C>>) {
    tokio::spawn(... async move {
        let futures = side_effects.iter()
            .filter(|se| se.enable(input.query.clone()))
            .map(|se| se.run(input.clone()));
        let _ = join_all(futures).await;
    });
}

A fragment of candidate_pipeline.rs · code by X, Apache 2.0, commit 28e414f

The key things here are tokio::spawn and let _ =. The task is started in the background, its result is neither awaited nor checked. The answer to the user has already been sent.

What the side effects do in the feed: record which posts were shown (so as not to show them again), update the cache of ranked posts, emit events about the ads, write statistics about the authors and the logs of the experiments.

All of them must be off the critical path. Recording the impressions may take tens of milliseconds — making the user wait for that, for the sake of something only the next request needs, makes no sense.

Common mistakes and hidden rocks

Interview questions

One-screen cheat sheet

Primary sources

• candidate_pipeline.rs — the order of the stages, the parallelism, the error handling.
• filter.rs, source.rs and their neighbours — the definitions of the stage types.
• phoenix_candidate_pipeline.rs — all 75 stages of the feed in order.
• The course: the chapter «The architecture of the runtime» on the runtime architecture and resilience, the chapter «The multi-stage funnel» on multi-stage design, the chapter «How to design a system from scratch» on layers with different goals.