02. Recommendation / Ranking System — Low-Level Design¶

~20 min read · Part 3 of 4 (Overview → HLD → LLD → Q&A)

The HLD named the boxes. This file opens the four that carry the design's weight — candidate generation, ranking, the feature pipeline, and the feedback loop — and pins down the schemas, the retrieve-then-rank algorithm as it burns the 100 ms budget, the sequence of a single homepage render, and the concurrency corners where a recommender quietly corrupts itself.

Data models¶

Feature store (online, read on every request)¶

Features are keyed by entity so the orchestrator can batch a multi-get for the user plus ~500 candidate items in one round trip.

KEY: user:{user_id}
VALUE (packed, ~1 KB):
  emb_v         : int8[256]        # user embedding, used for ANN retrieval
  taste_batch   : float[64]        # long-term genre/attribute affinities (daily batch)
  recent_items  : [item_id × 20]   # last N played, STREAMING-updated (seconds fresh)
  session_ctx   : {device, hour_of_day, session_started_at}
  updated_batch : ts               # when the batch features were written
  updated_stream: ts               # when the streaming features were written

KEY: item:{item_id}
VALUE (~1 KB):
  emb_v         : int8[256]        # item embedding (ANN index is built from these)
  meta          : {genres[], year, runtime, maturity, lang}
  stats_batch   : {ctr_30d, plays_30d, avg_completion}   # daily
  stats_stream  : {trending_10m, plays_1h}               # streaming, short window

KEY: cf:{user_id}                  # precomputed candidate pool
VALUE:
  neighbors     : [item_id × 500]  # collaborative-filtering nominees, daily batch
  built_at      : ts

Two non-obvious choices. Splitting each entity's features into *_batch and *_stream fields with separate updated_* timestamps is deliberate: the two are written by different pipelines at different cadences, and carrying both timestamps lets the ranker (and on-call) tell "the model is bad" apart from "the streaming features are stale." Storing recent_items denormalized on the user record — rather than joining a plays table at request time — is what makes recency a single key read inside the 20 ms feature-fetch slice instead of a query.

Impression / interaction log (feedback rail, append-only)¶

event {
  request_id   # joins back to the exact serving decision + model_version
  user_id
  item_id
  row_id       # which row it appeared in
  position     # rank within the row  — REQUIRED for position-bias correction
  type         # impression | click | play | skip | thumb_up | thumb_down
  model_version
  is_exploration  # true if this slot was an exploration pick
  ts
}

position and is_exploration are the fields juniors omit and seniors insist on. Without position, training cannot separate "ranked #1 so it got clicked" from "genuinely loved," and the model learns to reward whatever it already puts on top — the presentation-bias trap. is_exploration lets training weight exploration impressions correctly instead of treating a random slot's outcome as a confident prediction's outcome.

Model registry¶

model {
  model_version, artifact_uri, feature_schema_hash,
  trained_at, offline_metrics:{auc, ndcg}, status: shadow|canary|prod
}

feature_schema_hash guards against the classic training/serving skew bug — the serving side refuses to load a model whose expected features do not match what the online store actually provides.

Component internals¶

Component 1 — Candidate generation (multi-source retrieval)¶

Responsibility: narrow 10M items to ~500 candidates per request, recall-oriented, inside 15 ms, tolerant of any one source failing.

class CandidateGenerator:
    def generate(user, ctx, deadline_ms=15) -> list[Candidate]:
        # Fan out to sources in PARALLEL, each with its own share of the deadline.
        futures = [
            ann.query(user.emb_v, k=300),          # embedding neighbors
            store.get(f"cf:{user.id}").neighbors,   # precomputed CF pool
            trending.top(ctx, k=100),               # popularity floor / cold-start
            realtime.similar_to(user.recent_items[:1], k=100),  # "because you just watched"
        ]
        results = gather(futures, timeout=deadline_ms)   # drop late sources
        return dedup_merge(results, cap=500)             # union, keep source tags

Each source carries a source tag through the merge so the orchestrator can build row themes later ("Because you watched X" comes from the real-time source; "Trending Now" from the trending pool). The ANN query is an approximate nearest-neighbor lookup over the in-memory item-embedding index — it returns ~300 items near the user embedding in a few milliseconds by trading a little recall for speed, which is exactly the right trade at the recall stage since ranking will re-order them anyway. The realtime.similar_to(recent_items[:1]) source is the load-bearing piece for the threaded scenario: it takes the item the user finished seconds ago and pulls its neighbor list, so the just-finished crime drama seeds crime-drama candidates on the very next load without waiting for any batch job.

Component 2 — Ranking service (batched scoring)¶

Responsibility: score all ~500 candidates in one batched forward pass, ~40 ms, and stay attributable to a model version.

class RankingService:
    def rank(user, candidates, ctx) -> list[Scored]:
        # ONE batched feature fetch for user + all candidate items.
        feats = feature_store.mget([f"user:{user.id}"] +
                                   [f"item:{c.id}" for c in candidates])
        # Assemble a [N × F] matrix; impute missing features rather than error.
        X = build_matrix(user, candidates, feats, ctx)      # N≈500 rows
        scores = model.predict_batch(X)                     # single forward pass
        return [Scored(c.id, s, c.source) for c, s in zip(candidates, scores)]

The batched multi-get and the batched inference are the two moves that keep this within budget: fetching 500 keys in one round trip amortizes network cost, and scoring 500 rows in one forward pass amortizes model-load and kernel-launch cost. Scoring cost is linear in N, which is precisely why candidate generation caps the set at 500 — the ranker's latency and the size of the inference fleet both scale with N, so recall is bought only up to the point ranking quality stops improving. build_matrix imputes missing features (a fresh item with no stats_stream) rather than throwing, so a partially-cold candidate is scored conservatively instead of crashing the request.

Component 3 — Feature pipeline (batch + streaming dual-write)¶

Responsibility: keep the online store fresh at two cadences without either pipeline blocking serving.

# Streaming path (Flink), consuming the event bus continuously:
def on_event(e):                       # e: a play/click/impression
    if e.type == "play":
        redis.lpush_capped(f"user:{e.user_id}.recent_items", e.item_id, cap=20)
        redis.set_field(f"user:{e.user_id}.updated_stream", now())
        counters.incr(f"item:{e.item_id}.plays_1h", window="1h")   # sliding window

# Batch path (daily), writing slow features:
def nightly_backfill():
    for user in users:
        store.merge(f"user:{user.id}", {
            "taste_batch": recompute_taste(user),
            "emb_v":       embedding_model.encode(user),
            "updated_batch": now(),
        })

The two paths write disjoint fields of the same key, which is why the dual-write does not race on the value — streaming touches recent_items and *_stream, batch touches taste_batch, emb_v, and *_batch. lpush_capped keeps recent_items bounded at 20 so the user record stays ~1 KB and the read stays inside its slice. This is the component that pays for "reflect recent activity": a play event becomes a mutated recent_items within seconds, and the next homepage load reads it as an ordinary feature.

Component 4 — Feedback loop (logging + exploration)¶

Responsibility: log what was shown truthfully and inject exploration, so the training set does not collapse.

def shape_rows(scored, ctx) -> list[Row]:
    rows = group_into_rows(scored, ctx)          # by source tag / theme
    rows = diversify(rows)                        # cap per-genre dominance
    rows = apply_business_rules(rows)             # contractual promotes, maturity
    rows = inject_exploration(rows, epsilon=0.08) # ~8% of slots → high-uncertainty items
    return rows

def inject_exploration(rows, epsilon):
    # Thompson-style: sample some slots from items with high score-variance,
    # tag them is_exploration=True so the feedback log can weight them correctly.
    ...

inject_exploration spends ~8% of slots on items the model is uncertain about rather than most confident about. That costs a small, measurable dip in immediate click-through, and buys the exploration data that keeps the model from starving. The paired discipline lives in the client and gateway: every rendered item emits an impression with its position, so shown-and-ignored becomes a training negative — the difference between a model that learns taste and one that learns "keep showing whatever I already show."

Core algorithm — retrieve-then-rank under the 100 ms budget¶

Walk the threaded scenario through the pipeline: a user among the 200M finishes a crime drama, hits home, and the page must render in under 100 ms reflecting that finish.

Gateway (t=0 ms). Authenticate, mint request_id, cache-miss, forward to orchestrator.
Candidate fan-out (budget 15 ms). Four sources fire in parallel. ANN returns ~300 items near the user embedding. CF pool returns its precomputed 500 (capped later). Trending returns ~100. The real-time source reads user.recent_items[0] — the crime drama, written by the stream processor seconds ago — and pulls its ~100 similar items. Union and dedup to ~500 candidates, each source-tagged. If ANN misses its slice, it is dropped and the page proceeds on the other three.
Batched feature fetch (budget 20 ms). One multi-get for user:U plus 500 item:* keys across the sharded store. The user record already carries recent_items and updated_stream from the finish, so recency is in hand with no special-case code.
Batched scoring (budget 40 ms). Assemble the [500 × F] matrix and run one forward pass → 500 scores. The just-finished title, if it slipped into candidates, scores low because a "completed today" feature suppresses it; its neighbors score high because taste and recency features align. Cost here is 500 scores; multiplied across the fleet this is the 20,000 × 500 = 10M scores/second peak the ranking tier is sized for.
Blend and shape (budget 15 ms). Group by source tag into themed rows ("Because you watched…", "Trending", "Top Picks"), diversify so crime drama does not swamp all 40 rows, apply business rules, and swap ~8% of slots for exploration picks tagged is_exploration.
Return (~90 ms elapsed, ~10 ms network headroom). Rows go back with reasons and model_version; as the user scrolls, impression events fire onto the feedback rail, and one of them will be the impression of the crime-drama neighbors — closing the loop.

The budget is spent on the slowest branch per stage, not the sum of every call, which is why fan-out and batching appear at every step. Double the candidate count to 1,000 and step 4 alone would blow past 40 ms and the whole 100 ms budget — the reason recall is capped, not maximized.

Sequence diagram — one homepage render¶

Client        Gateway     Orchestrator    CandGen    FeatureStore    Ranker
  │ GET /home    │              │             │            │            │
  ├─────────────▶│  request_id  │             │            │            │
  │              ├─────────────▶│             │            │            │
  │              │              ├── generate ▶│            │            │
  │              │              │   (ANN,CF,trending,realtime in ∥)     │
  │              │              │◀ ~500 cands─┤            │            │
  │              │              ├── mget(user + 500 items) ▶│           │
  │              │              │◀──── features ───────────┤            │
  │              │              ├────────── rank(cands, feats) ────────▶│
  │              │              │◀──────────── 500 scores ──────────────┤
  │              │              ├─ blend/diversify/explore │            │
  │              │◀── rows ──────┤             │            │            │
  │◀─── rows ────┤              │             │            │            │
  │  (scroll → fire impressions async to Event Bus, tagged request_id)  │
  ├───────────────────────────── POST /feedback ──────────────────────▶ (Event Bus)

One parallel fan-out, one batched feature read, one batched inference, one shaping pass; feedback leaves on a separate async rail after the response.

Concurrency and edge cases¶

Batch vs streaming write race. The two feature pipelines write the same key concurrently. Resolved by having them own disjoint fields (streaming: recent_items, *_stream; batch: taste_batch, emb_v, *_batch) so there is no last-writer-wins conflict on shared state; each path uses field-level updates, not a whole-value overwrite.
Read-your-writes on recency. After a play, the user expects the next load to reflect it. The stream processor's write to recent_items must land before that next request reads it. This is eventually consistent with a lag of seconds; the real-time candidate source plus request-time features hedge the gap so the just-finished item influences ranking even if the streaming aggregate is a beat behind.
Impression/click join races. An impression and its click arrive as separate events, possibly out of order, and the click's request_id ties them together. Training joins on request_id in the warehouse, not in real time, so ordering on the bus does not matter — the append-only log is the source of truth.
Duplicate feedback events. Client retries can double-log an impression. Dedup on (request_id, item_id, type, ts) at warehouse load; slightly over-counting impressions is harmless to features but corrected before training.
Idempotent model rollout. Promoting a model is a registry pointer swap; the feature_schema_hash check makes loading idempotent and refuses a model that expects features the store does not have, preventing silent training/serving skew.
Partial candidate/feature failure. A dropped ANN source or a missing feature must degrade, not error: the orchestrator proceeds with fewer candidates, and the ranker imputes missing features. The page always renders — worse, never blank.
Exploration double-counting. An exploration slot's outcome must not be read as a confident prediction's outcome; the is_exploration flag lets training down-weight or separately model those events, or the bandit's own reward accounting inflates and the exploration policy destabilizes.