Reddit's actual interview questions

A thorough pass through the 1Point3Acres (一亩三分地) Chinese forum 面经 (logged-in, May 2026): ~35 Reddit threads mapped, ~30 read in full incl. replies, spanning phone screens, full onsites, MLE/ML-infra/ML-platform/Staff-MLE/Senior-DS reports from 2021–2026. They independently and verbatim cross-validate the darkinterview set and reveal the senior/staff loop structure you won't get from a question list. Candidate-reported, points-gated, translated — high-probability intel, not gospel.

New senior/staff questions the forum revealed (beyond the 19)

Round structure by track + org / comp intel

• Process: recruiter screen tells you exactly what each round covers; ~4-week timeline; Hiring Committee (HC) review then offer call ~1–2 days later. Small company → "team-based hiring," no formal team-match (you can pass and still not place — happened to a senior candidate who matched 4 orgs and got none).
• Orgs hiring: Search / Recommendations (openly knows its RecSys is weak, hiring senior/staff aggressively, pays up; → Reddit Answers RAG) and Ads (lots of HC; rougher tooling, more political). If your background is ranking/retrieval, target Search/Rec and mirror it in your deep-dive.
• Comp / structure: Staff MLE ~\$700K remote; Senior ~\$350–400K; IC3 ~\$245K base + \$85K RSU. RSU vests fully within year 1, quarterly; year-2+ is performance-based refresh (stable; "normal performance ≈ matches last year"). Fully remote (incl. outside the US, not China); layoff risk currently low.
• Open ML question raised in the org AMA: is Ads ranking on a DNN yet or still tree models? — be ready to discuss DNN vs GBDT for ad CTR.

Faithful translations of actual candidate 面经 (interview write-ups) from the 1Point3Acres forum, round-by-round, with outcomes. These are the real, detailed loops — not a summary. Candidate-reported and translated from Chinese; orgs and levels are as the posters stated.

Full problem statements with worked Python solutions and follow-ups. Reddit leans toward practical / OOD / multi-part coding (Tennis, Report Chain, Moderator List, Billing Replay) plus a few classic LeetCode (Word Search, Word Ladder, Shortest Palindrome, Odd-Even List). Drill the matching patterns in the Coding guide.

Problem Overview

Design a two-player game with three core methods:

• add_score(player)
• get_score()
• get_result()

The interview then adds a follow-up: convert the score into a human-readable tennis-style format such as love, 15-30, and deuce.

In Part 1, scores are stored as raw point counts. In Part 2, those raw counts are rendered using standard tennis terminology.

This is the classic tennis scoring question with one explicit normalization rule:

• A player must lead by 2 points to win.
• If the game reaches a tie after 3-3, collapse it back to 3-3 immediately.
• 4-4 becomes 3-3.
• That means later tied states such as 5-5 never persist as observable scores.

You can assume there are exactly two players.

Part 1: Numeric Game API

Problem Statement

Implement a Game class with these methods:

class Game:
    def add_score(self, player: str) -> None:
        pass

    def get_score(self) -> tuple[int, int]:
        pass

    def get_result(self) -> str | None:
        pass

Requirements

• Only two players exist.
• add_score(player) increments the specified player's score.
• get_score() returns the current numeric score for both players.
• get_result() returns the winner once someone is ahead by at least 2 points and has at least 4 points.
• When the score becomes tied after 3-3, normalize it back to 3-3 immediately.

Example

game = Game("player1", "player2")

game.add_score("player1")  # 1-0
game.add_score("player2")  # 1-1
game.add_score("player1")  # 2-1
game.add_score("player1")  # 3-1

game.get_score()   # (3, 1)
game.get_result()  # None

game.add_score("player2")  # 3-2
game.add_score("player2")  # 3-3
game.add_score("player1")  # 4-3
game.add_score("player2")  # 4-4 -> normalize to 3-3

game.get_score()   # (3, 3)
game.get_result()  # None

Part 1 Solution

class Game:
    def __init__(self, player1: str, player2: str):
        self.player1 = player1
        self.player2 = player2
        self.scores = {
            player1: 0,
            player2: 0,
        }

    def add_score(self, player: str) -> None:
        if player not in self.scores:
            raise ValueError(f"Unknown player: {player}")

        if self.get_result() is not None:
            raise ValueError("Game already has a winner")

        self.scores[player] += 1
        self._normalize_deuce()

    def get_score(self) -> tuple[int, int]:
        return (self.scores[self.player1], self.scores[self.player2])

    def get_result(self) -> str | None:
        p1 = self.scores[self.player1]
        p2 = self.scores[self.player2]

        if max(p1, p2) >= 4 and abs(p1 - p2) >= 2:
            return self.player1 if p1 > p2 else self.player2

        return None

    def _normalize_deuce(self) -> None:
        p1 = self.scores[self.player1]
        p2 = self.scores[self.player2]

        if p1 >= 4 and p1 == p2:
            self.scores[self.player1] = 3
            self.scores[self.player2] = 3

Complexity

• add_score: O(1)
• get_score: O(1)
• get_result: O(1)
• Space: O(1)

Part 2: Match Wrapper + Human Score

Problem Statement

Now write a Match class that builds a Game object internally and exposes a human-readable score using tennis terms.

class Match:
    def __init__(self, player1: str, player2: str):
        pass

    def point_won_by(self, player: str) -> None:
        pass

    def score(self) -> str:
        pass

    def result(self) -> str | None:
        pass

Human Score Rules

• 0 maps to love
• 1 maps to 15
• 2 maps to 30
• 3 maps to 40
• Tied scores below 3-3 are rendered normally, such as love-love, 15-15, and 30-30
• Raw 3-3 corresponds to tennis 40-40, which is rendered as deuce
• Raw 4-3 or 3-4 is advantage <player>
• If a player wins, return winner <player>

Because tied scores above 3-3 collapse back to 3-3, deuce naturally reappears after advantage is lost.

Example

match = Match("alice", "bob")

match.point_won_by("alice")
match.score()   # "15-love"

match.point_won_by("bob")
match.point_won_by("alice")
match.point_won_by("bob")
match.point_won_by("alice")
match.point_won_by("bob")

match.score()   # "deuce"

match.point_won_by("alice")
match.score()   # "advantage alice"

match.point_won_by("bob")
match.score()   # "deuce"

match.point_won_by("bob")
match.score()   # "advantage bob"

match.point_won_by("bob")
match.result()  # "winner bob"

Part 2 Solution

class Match:
    SCORE_LABELS = {
        0: "love",
        1: "15",
        2: "30",
        3: "40",
    }

    def __init__(self, player1: str, player2: str):
        self.game = Game(player1, player2)

    def point_won_by(self, player: str) -> None:
        self.game.add_score(player)

    def score(self) -> str:
        winner = self.game.get_result()
        if winner is not None:
            return f"winner {winner}"

        p1, p2 = self.game.get_score()
        player1 = self.game.player1
        player2 = self.game.player2

        if p1 >= 3 and p2 >= 3:
            if p1 == p2:
                return "deuce"

            leader = player1 if p1 > p2 else player2
            return f"advantage {leader}"

        return f"{self.SCORE_LABELS[p1]}-{self.SCORE_LABELS[p2]}"

    def result(self) -> str | None:
        winner = self.game.get_result()
        if winner is None:
            return None
        return f"winner {winner}"

Source: darkinterview.com (real Reddit interview report, captured May 2026)

Problem Overview

You are given an org chart as a list[list[str]].

Each inner list has the form:

[manager, report_1, report_2, ...]

For example:

[
    ["A", "B", "C"],
    ["B", "E"],
    ["C", "D"],
]

means:

• A manages B and C
• B manages E
• C manages D

This produces the report chain:

A
....B
........E
....C
........D

This interview commonly has three main parts, and some interviewers add a fourth follow-up.

Assume:

• The input represents a valid management tree
• There is exactly one top-level manager
• Child order should match the order given in the input
• Four dots (....) represent one indentation level

Part 1: Print The Full Report Chain

Problem Statement

Build the org tree and print the full hierarchy using the indentation format above.

Example

relations = [
    ["A", "B", "C"],
    ["B", "E"],
    ["C", "D"],
]

Output:

A
....B
........E
....C
........D

Part 1 Solution

Build a children adjacency list and find the root, then do a DFS render.

from collections import defaultdict


def build_children(relations: list[list[str]]) -> tuple[dict[str, list[str]], str]:
    children: dict[str, list[str]] = defaultdict(list)
    parent: dict[str, str] = {}
    all_people = set()

    for row in relations:
        manager, *reports = row
        all_people.add(manager)
        children[manager]

        for report in reports:
            all_people.add(report)
            children[manager].append(report)
            children[report]
            parent[report] = manager

    roots = [person for person in all_people if person not in parent]
    if len(roots) != 1:
        raise ValueError("Expected exactly one root")

    return children, roots[0]


def render_full_chain(relations: list[list[str]]) -> str:
    children, root = build_children(relations)
    lines: list[str] = []

    def dfs(node: str, depth: int) -> None:
        lines.append(f"{'....' * depth}{node}")
        for child in children[node]:
            dfs(child, depth + 1)

    dfs(root, 0)
    return "\n".join(lines)

Part 2: Emit All Skip-Level Pairs

Problem Statement

As a manager of managers, you may want to hold skip-level meetings with employees who are exactly two levels below you.

Return all valid (manager, employee) skip-level pairs.

Using the same tree:

• A can skip over B to meet E
• A can skip over C to meet D

Output can be returned in any convenient format as long as the pairs are correct.

Example

[("A", "E"), ("A", "D")]

Part 2 Solution

Once you have the tree, every valid skip-level pair is just a (manager, grandchild) relationship.

def all_skip_level_pairs(relations: list[list[str]]) -> list[tuple[str, str]]:
    children, root = build_children(relations)
    pairs: list[tuple[str, str]] = []

    def dfs(node: str) -> None:
        for child in children[node]:
            for grandchild in children[child]:
                pairs.append((node, grandchild))
            dfs(child)

    dfs(root)
    return pairs

Part 3: Given A Person, Print Only Their Chain

Problem Statement

Given a target employee, print:

• the single management path from the root down to that employee
• all descendants under that target employee

The output must use the same indentation format as Part 1.

Example

For target B, using:

relations = [
    ["A", "B", "C"],
    ["B", "E"],
    ["C", "D"],
]

output:

A
....B
........E

If the target is C, the output is:

A
....C
........D

Part 3 Solution

This is the part that usually trips people up. You need both:

• the path from the root down to the target
• the full subtree under the target

The cleanest way is to keep a parent map so you can reconstruct the upward chain, then render only the target's descendants afterward.

The key detail is that the target should appear exactly once:

• print the target as part of the root-to-target path
• then DFS only the target's children, not the target again

from collections import defaultdict


def build_graph(
    relations: list[list[str]],
) -> tuple[dict[str, list[str]], dict[str, str], str]:
    children: dict[str, list[str]] = defaultdict(list)
    parent: dict[str, str] = {}
    all_people = set()

    for row in relations:
        manager, *reports = row
        all_people.add(manager)
        children[manager]

        for report in reports:
            all_people.add(report)
            children[manager].append(report)
            children[report]
            parent[report] = manager

    roots = [person for person in all_people if person not in parent]
    if len(roots) != 1:
        raise ValueError("Expected exactly one root")

    return children, parent, roots[0]


def render_chain_for(relations: list[list[str]], target: str) -> str:
    children, parent, _root = build_graph(relations)
    if target not in children:
        raise ValueError(f"Unknown employee: {target}")

    path = [target]
    current = target
    while current in parent:
        current = parent[current]
        path.append(current)
    path.reverse()

    lines: list[str] = []
    for depth, name in enumerate(path):
        lines.append(f"{'....' * depth}{name}")

    def dfs_subtree(node: str, depth: int) -> None:
        lines.append(f"{'....' * depth}{node}")
        for child in children[node]:
            dfs_subtree(child, depth + 1)

    for child in children[target]:
        dfs_subtree(child, len(path))

    return "\n".join(lines)

Follow-Up: Lowest Common Manager

Some interviewers extend the problem and ask:

Given two employees, return their lowest common manager.

For the same tree:

lowest_common_manager("C", "E") == "A"

Follow-Up Solution

The parent map also makes the lowest-common-manager follow-up straightforward:

def lowest_common_manager(
    relations: list[list[str]], employee1: str, employee2: str
) -> str:
    children, parent, _root = build_graph(relations)
    if employee1 not in children or employee2 not in children:
        raise ValueError("Unknown employee")

    ancestors = set()
    current = employee1
    ancestors.add(current)

    while current in parent:
        current = parent[current]
        ancestors.add(current)

    current = employee2
    while current not in ancestors:
        current = parent[current]

    return current

Full Reference Implementation

If the interviewer keeps adding follow-ups, it is usually better to unify everything into one reusable OrgChart class.

from collections import defaultdict


class OrgChart:
    INDENT = "...."

    def __init__(self, relations: list[list[str]]):
        self.children: dict[str, list[str]] = defaultdict(list)
        self.parent: dict[str, str] = {}
        self._appearance_order: list[str] = []
        seen = set()

        for row in relations:
            if not row:
                continue

            manager, *reports = row
            self._remember(manager, seen)
            self.children[manager]

            for report in reports:
                self._remember(report, seen)
                self.children[manager].append(report)
                self.children[report]
                self.parent[report] = manager

        roots = [name for name in self._appearance_order if name not in self.parent]
        if len(roots) != 1:
            raise ValueError("Input must contain exactly one root")

        self.root = roots[0]

    def _remember(self, name: str, seen: set[str]) -> None:
        if name not in seen:
            seen.add(name)
            self._appearance_order.append(name)

    def render_full_chain(self) -> str:
        lines: list[str] = []
        self._render_subtree(self.root, 0, lines)
        return "\n".join(lines)

    def all_skip_level_pairs(self) -> list[tuple[str, str]]:
        pairs: list[tuple[str, str]] = []

        def dfs(node: str) -> None:
            for child in self.children[node]:
                for grandchild in self.children[child]:
                    pairs.append((node, grandchild))
                dfs(child)

        dfs(self.root)
        return pairs

    def render_chain_for(self, target: str) -> str:
        if target not in self.children:
            raise ValueError(f"Unknown employee: {target}")

        path = self._path_from_root(target)
        lines: list[str] = []

        for depth, name in enumerate(path):
            lines.append(f"{self.INDENT * depth}{name}")

        target_depth = len(path) - 1
        for child in self.children[target]:
            self._render_subtree(child, target_depth + 1, lines)

        return "\n".join(lines)

    def lowest_common_manager(self, employee1: str, employee2: str) -> str:
        if employee1 not in self.children or employee2 not in self.children:
            raise ValueError("Unknown employee")

        ancestors = set()
        current = employee1
        ancestors.add(current)

        while current in self.parent:
            current = self.parent[current]
            ancestors.add(current)

        current = employee2
        while current not in ancestors:
            current = self.parent[current]

        return current

    def _path_from_root(self, target: str) -> list[str]:
        path = [target]
        current = target

        while current in self.parent:
            current = self.parent[current]
            path.append(current)

        path.reverse()
        return path

    def _render_subtree(self, node: str, depth: int, lines: list[str]) -> None:
        lines.append(f"{self.INDENT * depth}{node}")
        for child in self.children[node]:
            self._render_subtree(child, depth + 1, lines)

Example Usage

relations = [
    ["A", "B", "C"],
    ["B", "E"],
    ["C", "D"],
]

chart = OrgChart(relations)

print(chart.render_full_chain())
# A
# ....B
# ........E
# ....C
# ........D

print(chart.all_skip_level_pairs())
# [('A', 'E'), ('A', 'D')]

print(chart.render_chain_for("B"))
# A
# ....B
# ........E

print(chart.lowest_common_manager("C", "E"))
# A

Complexity

Let n be the number of employees.

• Build graph: O(n)
• Part 1 render: O(n)
• Part 2 skip-level pairs: O(n)
• Part 3 target chain render: O(h + s)
  • h is the height from root to target
  • s is the size of the target's subtree
• Lowest common manager: O(h)
• Space: O(n)

Source: darkinterview.com (real Reddit interview report, captured May 2026)

Problem Overview

Implement a Reddit-style moderator list from a newline-delimited log string.

The interview usually comes in three parts:

• A single mod list
• Multiple communities
• Demotions that reorder the current mod list

The core rule is:

• A moderator can remove another moderator only if they are above that moderator in the current mod list.
• Higher in the list means the moderator has an earlier effective access timestamp.

In the first part, each log line has four comma-separated fields:

target, action, actor, timestamp

• target: the moderator receiving the action
• action: add or remove in Parts 1 and 2, plus demote in Part 3
• actor: the moderator who performed the action
• timestamp: integer timestamp, increasing over time

Assume:

• The logs are already sorted by timestamp
• In Part 1, timestamps are unique
• In Parts 2 and 3, timestamps are unique within each community
• Every logged action is valid
• SYSTEM may appear as an actor to seed the first moderator, but SYSTEM is not part of the returned mod list unless explicitly added
• If a removed moderator is added again later, they get a new access timestamp

Each part asks for the same three operations:

• Constructor: build the state from the log string
• can_remove_mod(...)
• get_mod_list(...)

Part 1: Single Moderator List

Problem Statement

Implement a class for a single moderator list:

class ModList:
    def __init__(self, logs: str):
        pass

    def can_remove_mod(self, actor: str, target: str) -> bool:
        pass

    def get_mod_list(self) -> list[str]:
        pass

Rules

• add means target becomes an active moderator at timestamp
• remove means target is no longer an active moderator
• can_remove_mod(actor, target) returns True only if:
  • both moderators are currently active
  • actor != target
  • actor has an earlier effective access timestamp than target
• get_mod_list() returns active moderators from top to bottom of the mod list

Example

logs = """
alice,add,SYSTEM,1
bob,add,alice,2
carol,add,alice,3
dave,add,bob,4
carol,remove,alice,5
""".strip()

mod_list = ModList(logs)

mod_list.can_remove_mod("alice", "bob")   # True
mod_list.can_remove_mod("bob", "alice")   # False
mod_list.can_remove_mod("bob", "dave")    # True
mod_list.get_mod_list()                   # ["alice", "bob", "dave"]

Part 1 Solution

Replay the log once and store each active moderator's effective access timestamp.

class ModList:
    def __init__(self, logs: str):
        self.active_since: dict[str, int] = {}

        for raw_line in logs.splitlines():
            line = raw_line.strip()
            if not line:
                continue

            target, action, actor, ts = [part.strip() for part in line.split(",")]
            timestamp = int(ts)

            if action == "add":
                self.active_since[target] = timestamp
            elif action == "remove":
                self.active_since.pop(target, None)
            else:
                raise ValueError(f"Unsupported action: {action}")

    def can_remove_mod(self, actor: str, target: str) -> bool:
        if actor == target:
            return False
        if actor not in self.active_since or target not in self.active_since:
            return False
        return self.active_since[actor] < self.active_since[target]

    def get_mod_list(self) -> list[str]:
        return [
            mod
            for mod, _ in sorted(
                self.active_since.items(),
                key=lambda item: (item[1], item[0]),
            )
        ]

Complexity

• Constructor: O(n)
• can_remove_mod: O(1)
• get_mod_list: O(m log m), where m is the number of active moderators
• Space: O(m)

Part 2: Add Community Support

Problem Statement

Now extend the problem to multiple communities. Each log line becomes:

community, target, action, actor, timestamp

Implement:

class CommunityModList:
    def __init__(self, logs: str):
        pass

    def can_remove_mod(self, community: str, actor: str, target: str) -> bool:
        pass

    def get_mod_list(self, community: str) -> list[str]:
        pass

Rules

• Moderator membership is tracked independently for each community
• The same username can appear in multiple communities with different access timestamps
• can_remove_mod only considers the given community

Example

logs = """
python,alice,add,SYSTEM,1
python,bob,add,alice,2
python,carol,add,alice,3
datascience,maya,add,SYSTEM,4
datascience,bob,add,maya,5
python,carol,remove,alice,6
""".strip()

mod_list = CommunityModList(logs)

mod_list.get_mod_list("python")        # ["alice", "bob"]
mod_list.get_mod_list("datascience")   # ["maya", "bob"]
mod_list.can_remove_mod("python", "alice", "bob")        # True
mod_list.can_remove_mod("datascience", "bob", "maya")    # False

Part 2 Solution

Use a nested dictionary keyed by community, then by moderator.

from collections import defaultdict


class CommunityModList:
    def __init__(self, logs: str):
        self.active_since: dict[str, dict[str, int]] = defaultdict(dict)

        for raw_line in logs.splitlines():
            line = raw_line.strip()
            if not line:
                continue

            community, target, action, actor, ts = [
                part.strip() for part in line.split(",")
            ]
            timestamp = int(ts)

            if action == "add":
                self.active_since[community][target] = timestamp
            elif action == "remove":
                self.active_since[community].pop(target, None)
            else:
                raise ValueError(f"Unsupported action: {action}")

    def can_remove_mod(self, community: str, actor: str, target: str) -> bool:
        mods = self.active_since.get(community, {})

        if actor == target:
            return False
        if actor not in mods or target not in mods:
            return False
        return mods[actor] < mods[target]

    def get_mod_list(self, community: str) -> list[str]:
        mods = self.active_since.get(community, {})
        return [
            mod
            for mod, _ in sorted(
                mods.items(),
                key=lambda item: (item[1], item[0]),
            )
        ]

Complexity

• Constructor: O(n)
• can_remove_mod: O(1)
• get_mod_list(community): O(m log m)
• Space: O(total active moderators across all communities)

Part 3: Support Moderator Demotion

Problem Statement

Extend the multi-community version to support a third action:

community, target, demote, actor, timestamp

Demotion means the target moderator stays active, but their effective access timestamp becomes the demotion timestamp. In other words, they move down in that community's mod list.

Implement the same three functions:

class ReorderableCommunityModList:
    def __init__(self, logs: str):
        pass

    def can_remove_mod(self, community: str, actor: str, target: str) -> bool:
        pass

    def get_mod_list(self, community: str) -> list[str]:
        pass

Example

logs = """
python,alice,add,SYSTEM,1
python,bob,add,alice,2
python,carol,add,alice,3
python,bob,demote,alice,4
""".strip()

mod_list = ReorderableCommunityModList(logs)

mod_list.get_mod_list("python")                     # ["alice", "carol", "bob"]
mod_list.can_remove_mod("python", "carol", "bob")  # True
mod_list.can_remove_mod("python", "bob", "carol")  # False

Part 3 Solution

The only real change is handling demote by overwriting the target moderator's effective timestamp.

from collections import defaultdict


class ReorderableCommunityModList:
    def __init__(self, logs: str):
        self.active_since: dict[str, dict[str, int]] = defaultdict(dict)

        for raw_line in logs.splitlines():
            line = raw_line.strip()
            if not line:
                continue

            community, target, action, actor, ts = [
                part.strip() for part in line.split(",")
            ]
            timestamp = int(ts)
            mods = self.active_since[community]

            if action == "add":
                mods[target] = timestamp
            elif action == "remove":
                mods.pop(target, None)
            elif action == "demote":
                if target in mods:
                    mods[target] = timestamp
            else:
                raise ValueError(f"Unsupported action: {action}")

    def can_remove_mod(self, community: str, actor: str, target: str) -> bool:
        mods = self.active_since.get(community, {})

        if actor == target:
            return False
        if actor not in mods or target not in mods:
            return False
        return mods[actor] < mods[target]

    def get_mod_list(self, community: str) -> list[str]:
        mods = self.active_since.get(community, {})
        return [
            mod
            for mod, _ in sorted(
                mods.items(),
                key=lambda item: (item[1], item[0]),
            )
        ]

Follow-Up Discussion

• How would you support get_mod_list() in O(m) without sorting every time?
• How would you validate invalid logs instead of assuming all actions are valid?
• What if timestamps are not unique?
• How would you support querying historical mod-list state at a past timestamp?

Source: darkinterview.com (real Reddit interview report, captured May 2026)

class Logger:
    """
    Approach 1: simple hashmap storing next-allowed timestamp per message.

    Time:  O(1) average per call (dict lookup/insert)
    Space: O(M) where M = number of distinct messages ever seen — grows forever.
    """

    def __init__(self) -> None:
        # message -> earliest timestamp at which printing is again allowed
        self._next_allowed: dict[str, int] = {}

    def shouldPrintMessage(self, timestamp: int, message: str) -> bool:
        allowed_at = self._next_allowed.get(message, 0)
        if timestamp < allowed_at:
            return False
        # Allow and set the next allowed time
        self._next_allowed[message] = timestamp + 10
        return True


# ── driver ──────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    logger = Logger()
    calls = [
        (1,  "foo"),   # True  — first time
        (2,  "bar"),   # True  — first time
        (3,  "foo"),   # False — too soon (next allowed = 11)
        (8,  "bar"),   # False — too soon (next allowed = 12)
        (10, "foo"),   # False — 10 < 11
        (11, "foo"),   # True  — 11 >= 11, allowed; next = 21
    ]
    expected = [True, True, False, False, False, True]
    for (ts, msg), exp in zip(calls, expected):
        result = logger.shouldPrintMessage(ts, msg)
        status = "OK" if result == exp else "FAIL"
        print(f"t={ts:2d}  msg={msg!r:5s}  got={result}  exp={exp}  [{status}]")

from collections import deque


class LoggerWithEviction:
    """
    Approach 2: bounded memory via lazy eviction using a deque of (expire_time, message).

    Invariant: the deque is sorted by expire_time (naturally, since timestamps
    are non-decreasing and expire_time = timestamp + 10).

    Time:  O(1) amortized — each message is enqueued and dequeued at most once.
    Space: O(W) where W = max distinct messages in any 10-second window.
    """

    def __init__(self) -> None:
        self._allowed: dict[str, int] = {}          # msg -> next_allowed_time
        self._expiry_queue: deque[tuple[int, str]] = deque()  # (expire_at, msg)

    def _evict_expired(self, current_timestamp: int) -> None:
        """Remove messages whose rate-limit window has fully expired."""
        while self._expiry_queue:
            expire_at, msg = self._expiry_queue[0]
            if expire_at <= current_timestamp:
                # This entry is stale — but only delete from dict if the dict
                # still points to THIS expiry (a later print may have updated it).
                if self._allowed.get(msg) == expire_at:
                    del self._allowed[msg]
                self._expiry_queue.popleft()
            else:
                break  # queue is sorted; nothing further is expired

    def shouldPrintMessage(self, timestamp: int, message: str) -> bool:
        self._evict_expired(timestamp)

        allowed_at = self._allowed.get(message, 0)
        if timestamp < allowed_at:
            return False

        # Accept: record new expiry and push to queue
        new_expiry = timestamp + 10
        self._allowed[message] = new_expiry
        self._expiry_queue.append((new_expiry, message))
        return True


# ── driver ───────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    logger = LoggerWithEviction()
    calls = [
        (1,  "foo"),
        (2,  "bar"),
        (3,  "foo"),
        (8,  "bar"),
        (10, "foo"),
        (11, "foo"),
    ]
    expected = [True, True, False, False, False, True]
    for (ts, msg), exp in zip(calls, expected):
        result = logger.shouldPrintMessage(ts, msg)
        status = "OK" if result == exp else "FAIL"
        print(f"t={ts:2d}  msg={msg!r:5s}  got={result}  exp={exp}  [{status}]")

    # Demonstrate eviction: after t=20 all old entries should be gone
    logger.shouldPrintMessage(25, "baz")
    print(f"Dict size after t=25: {len(logger._allowed)}")  # should be 1 (just 'baz')

import threading


class ThreadSafeLogger:
    """
    Thread-safe Logger using a global RLock.

    Trade-off: global lock serializes all calls — easy to reason about,
    correct, but contended under high concurrency.

    For higher throughput: use a dict of per-key locks (key-sharded locking).
    """

    def __init__(self) -> None:
        self._next_allowed: dict[str, int] = {}
        self._lock = threading.RLock()  # RLock allows reentrant acquisition

    def shouldPrintMessage(self, timestamp: int, message: str) -> bool:
        with self._lock:
            allowed_at = self._next_allowed.get(message, 0)
            if timestamp < allowed_at:
                return False
            self._next_allowed[message] = timestamp + 10
            return True


class ShardedLockLogger:
    """
    Higher-throughput thread-safe Logger using N sharded locks.

    Messages are hashed into one of NUM_SHARDS buckets. Concurrent calls
    for messages in different shards proceed in parallel; only same-shard
    calls contend.

    Time:  O(1) per call
    Space: O(M) for the dict + O(NUM_SHARDS) for the locks
    """

    NUM_SHARDS = 64

    def __init__(self) -> None:
        self._next_allowed: dict[str, int] = {}
        self._locks = [threading.Lock() for _ in range(self.NUM_SHARDS)]

    def _shard(self, message: str) -> int:
        return hash(message) % self.NUM_SHARDS

    def shouldPrintMessage(self, timestamp: int, message: str) -> bool:
        lock = self._locks[self._shard(message)]
        with lock:
            allowed_at = self._next_allowed.get(message, 0)
            if timestamp < allowed_at:
                return False
            self._next_allowed[message] = timestamp + 10
            return True


# ── concurrent stress test ───────────────────────────────────────────────────
if __name__ == "__main__":
    import random

    logger = ShardedLockLogger()
    results = []
    errors = []

    def worker(tid: int) -> None:
        for i in range(100):
            ts = random.randint(1, 50)
            msg = random.choice(["foo", "bar", "baz"])
            try:
                r = logger.shouldPrintMessage(ts, msg)
                results.append(r)
            except Exception as e:
                errors.append(e)

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(20)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    print(f"Total calls: {len(results)}, errors: {len(errors)}")

"""
Rate Limiting Algorithm Taxonomy
=================================

1. FIXED WINDOW COUNTER
   - Divide time into fixed buckets (e.g., [0,10), [10,20), ...).
   - Count events per bucket; reject if count exceeds limit.
   - Pro: trivial, O(1) time and space per key.
   - Con: boundary burst — a user can fire 2x the limit by straddling
     two windows (N at t=9.9, N at t=10.1).

2. SLIDING WINDOW LOG (what LC 359 essentially is for 1-request limit)
   - Store the exact timestamps of all recent events in a sorted list.
   - On each request, evict entries older than (now - window) then check count.
   - Pro: perfectly accurate sliding window.
   - Con: O(N) space per key (N = requests in window); expensive to purge.

3. SLIDING WINDOW COUNTER (hybrid, used in practice)
   - Keep counters for current and previous fixed windows.
   - Estimate current sliding count = prev_count * overlap_ratio + curr_count.
   - Pro: O(1) space, very close to sliding accuracy.
   - Con: estimate, not exact; off by at most ~0.003% in practice.

4. TOKEN BUCKET
   - Each key has a bucket of capacity C that refills at rate R tokens/second.
   - A request costs 1 token; if bucket is empty, reject.
   - Pro: allows bursting up to C; smooth average rate R.
   - Con: state per key (current tokens, last refill time).

5. LEAKY BUCKET
   - Requests enter a queue (bucket); a worker drains at fixed rate R/sec.
   - Pro: perfectly smooth output, no bursting.
   - Con: adds latency; a steady stream at 2R causes queue to grow unbounded.
"""

import time


class TokenBucketRateLimiter:
    """
    Per-key token bucket rate limiter.

    capacity: maximum burst
    refill_rate: tokens added per second
    """

    def __init__(self, capacity: float, refill_rate: float) -> None:
        self._capacity = capacity
        self._refill_rate = refill_rate
        self._buckets: dict[str, tuple[float, float]] = {}
        # key -> (current_tokens, last_refill_wall_time)

    def allow(self, key: str) -> bool:
        now = time.monotonic()
        tokens, last_time = self._buckets.get(key, (self._capacity, now))

        # Refill tokens based on elapsed time
        elapsed = now - last_time
        tokens = min(self._capacity, tokens + elapsed * self._refill_rate)

        if tokens < 1.0:
            self._buckets[key] = (tokens, now)
            return False

        self._buckets[key] = (tokens - 1.0, now)
        return True


class SlidingWindowLogRateLimiter:
    """
    Sliding window log: exact but O(N) space per key.

    window_seconds: length of the sliding window
    max_requests: allowed requests in that window
    """

    def __init__(self, window_seconds: int, max_requests: int) -> None:
        from collections import deque
        self._window = window_seconds
        self._limit = max_requests
        self._logs: dict[str, deque] = {}

    def allow(self, key: str, timestamp: int) -> bool:
        from collections import deque
        if key not in self._logs:
            self._logs[key] = deque()
        log = self._logs[key]

        # Evict requests outside the window
        cutoff = timestamp - self._window
        while log and log[0] <= cutoff:
            log.popleft()

        if len(log) < self._limit:
            log.append(timestamp)
            return True
        return False


# ── demo ─────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    tb = TokenBucketRateLimiter(capacity=3, refill_rate=1)
    for i in range(5):
        print(f"Token bucket request {i+1}: {tb.allow('user_A')}")
    # First 3 allowed (burst), 4th and 5th rejected (no tokens)

    sw = SlidingWindowLogRateLimiter(window_seconds=10, max_requests=2)
    for ts in [1, 5, 9, 11, 15]:
        print(f"Sliding window t={ts}: {sw.allow('user_B', ts)}")
    # t=1: True, t=5: True, t=9: False (2 in [0,9]), t=11: True (evicts t=1), t=15: True

"""
Distributed Logger Rate Limiter using Redis.

Pattern A: Redis SET with NX + EXPIRY (simplest, best for this exact problem)
  - SET msg timestamp NX EX 10
  - NX = only set if key does not exist
  - EX 10 = auto-expire after 10 seconds
  - Atomic: no race conditions.

Pattern B: Redis sorted set sliding window (for N-per-window, not just dedup)
  - Key = "ratelimit:{user_id}"
  - ZADD key timestamp timestamp   (score=timestamp, member=timestamp)
  - ZREMRANGEBYSCORE key -inf (now - window)
  - ZCARD key to check count
  - Wrap in MULTI/EXEC for atomicity, or use a Lua script.

Pattern C: Redis + Lua script (atomic sliding window)
  - Lua executes atomically on the Redis server — no round trips, no races.
"""

# ── Pattern A: SET NX EX (pseudo-code with redis-py) ─────────────────────────
#
# import redis
# r = redis.Redis(host="localhost", port=6379)
#
# def should_print_message_distributed(timestamp: int, message: str) -> bool:
#     key = f"logger:{message}"
#     # SET key value NX EX seconds
#     # Returns True if key was newly set, None if it already existed
#     result = r.set(key, timestamp, nx=True, ex=10)
#     return result is not None
#
# Limitation: Redis TTL is wall-clock seconds, not logical timestamps.
# For production log dedup where "timestamp" is logical, use Pattern B/C.


# ── Pattern B: Sorted-set sliding window (simulated, no real Redis) ──────────

from collections import defaultdict
import bisect


class FakeRedis:
    """Minimal sorted-set simulation for interview illustration."""

    def __init__(self) -> None:
        self._zsets: dict[str, list[float]] = defaultdict(list)

    def zadd(self, key: str, score: float) -> None:
        bisect.insort(self._zsets[key], score)

    def zremrangebyscore(self, key: str, min_score: float, max_score: float) -> None:
        zset = self._zsets[key]
        lo = bisect.bisect_left(zset, min_score)
        hi = bisect.bisect_right(zset, max_score)
        del zset[lo:hi]

    def zcard(self, key: str) -> int:
        return len(self._zsets[key])


class DistributedSlidingWindowLogger:
    """
    Distributed rate limiter using a sorted-set sliding window in Redis.
    Each logical-timestamp unit represents 1 second.

    window: size of the rate-limit window in seconds
    max_count: max allowed prints in that window (1 for dedup)
    """

    def __init__(self, window: int = 10, max_count: int = 1) -> None:
        self._redis = FakeRedis()
        self._window = window
        self._max_count = max_count

    def shouldPrintMessage(self, timestamp: int, message: str) -> bool:
        key = f"logger:{message}"

        # 1. Evict entries older than the window
        cutoff = timestamp - self._window  # exclusive lower bound
        self._redis.zremrangebyscore(key, float("-inf"), cutoff)

        # 2. Count remaining entries
        count = self._redis.zcard(key)
        if count >= self._max_count:
            return False

        # 3. Add current timestamp (in real Redis this + steps 1-2 wrapped in Lua)
        self._redis.zadd(key, float(timestamp))
        return True


# ── Lua script for atomic execution on real Redis ───────────────────────────
SLIDING_WINDOW_LUA = """
local key      = KEYS[1]
local now      = tonumber(ARGV[1])
local window   = tonumber(ARGV[2])
local limit    = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', key, '-inf', now - window)
local count = redis.call('ZCARD', key)
if count < limit then
    redis.call('ZADD', key, now, now)
    redis.call('EXPIRE', key, window)
    return 1
end
return 0
"""
# Usage with redis-py:
# script = r.register_script(SLIDING_WINDOW_LUA)
# allowed = script(keys=[f"logger:{message}"], args=[timestamp, 10, 1])


# ── driver ───────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    logger = DistributedSlidingWindowLogger(window=10, max_count=1)
    calls = [(1, "foo"), (2, "bar"), (3, "foo"), (8, "bar"), (10, "foo"), (11, "foo")]
    expected = [True, True, False, False, False, True]
    for (ts, msg), exp in zip(calls, expected):
        result = logger.shouldPrintMessage(ts, msg)
        status = "OK" if result == exp else "FAIL"
        print(f"t={ts:2d}  msg={msg!r:5s}  got={result}  exp={exp}  [{status}]")

Source: darkinterview.com (real Reddit interview report, captured May 2026)

Problem Overview

You are given an existing API:

def get_chat_messages(message_id: int) -> list[Message]:
    ...

For a given message_id, it returns the surrounding chat context:

• up to 5 messages before the target
• the target message itself
• up to 5 messages after the target

Assume the returned list is already sorted by increasing message.id.

Now implement:

def merge_messages(ids: list[int]) -> list[Message]:
    ...

Given a list of message IDs, call get_chat_messages(...) for each ID, merge all returned windows, remove duplicates, and return the final messages in order.

The interviewer hint is important:

• do not solve this by dumping everything into a set and sorting afterward
• instead, use the fact that each API response is already ordered

Assume:

• Message.id is unique
• ids may contain duplicates
• windows may overlap heavily
• at the beginning or end of a conversation, fewer than 11 messages may be returned

Part 1: Merge Ordered Message Windows

Problem Statement. Implement merge_messages(ids) and return the merged, deduplicated message list in increasing ID order.

Example. If:

ids = [1, 3, 5]

and each call returns the surrounding context, then the merged result should be:

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

because the windows overlap and should collapse into one ordered result.

Part 1 Solution

The intended approach is a merge-style scan, not dedupe-then-sort.

• Sort the requested IDs.
• Fetch each ordered window.
• Maintain the largest message ID already appended.
• While scanning the next window, skip any message whose ID is already covered.

Because each window is sorted, once msg.id > last_id, everything after it in that same window is also in the correct order.

from dataclasses import dataclass


@dataclass(frozen=True)
class Message:
    id: int
    text: str


def merge_messages(ids: list[int]) -> list[Message]:
    if not ids:
        return []

    merged: list[Message] = []
    last_id: int | None = None

    for message_id in sorted(ids):
        window = get_chat_messages(message_id)

        for message in window:
            if last_id is not None and message.id <= last_id:
                continue

            merged.append(message)
            last_id = message.id

    return merged

Why This Works

• Sorting ids ensures we process earlier windows before later ones.
• Every get_chat_messages(...) response is already sorted.
• Overlap only appears as a prefix of a later window.
• So a single last_id check is enough to remove duplicates while preserving order.

This is the key insight the interviewer usually wants instead of:

• gather all messages
• deduplicate with set
• sort everything again

Complexity

• Sorting IDs: O(k log k), where k = len(ids)
• Scanning returned messages: O(T), where T is the total number of messages returned across all API calls
• Extra space: O(R) for the output

Since each API call returns at most 11 messages, T is at most 11k.

Part 2: Follow-Up If Messages Can Be Edited

Problem Statement. Suppose messages are editable and you now need to preserve version history instead of overwriting content in place. How would you model that?

Example. Suppose message 42 is created and later edited twice:

message = VersionedMessage.create(
    message_id=42,
    content="hello",
    created_at=100,
)

message.edit("hello world", edited_at=120)
message.edit("hello world!", edited_at=150)

message.latest().content    # "hello world!"
message.as_of(110).content  # "hello"
message.as_of(130).content  # "hello world"

If get_chat_messages(42, as_of=130) includes message 42, it should return the version whose content is "hello world", not the latest edit from time 150.

Part 2 Solution

The main design change is to separate:

• the stable logical message identity
• the immutable versions of that message over time

That means:

• message_id stays constant for the life of the message
• each edit creates a new version record
• the message tracks which version is current for fast latest reads
• edits are appended in chronological order so snapshot lookup can binary search version history

from dataclasses import dataclass, field


@dataclass(frozen=True)
class MessageVersion:
    version: int
    content: str
    edited_at: int


@dataclass
class VersionedMessage:
    id: int
    current_version: int
    versions: list[MessageVersion] = field(default_factory=list)

    @classmethod
    def create(cls, message_id: int, content: str, created_at: int) -> "VersionedMessage":
        return cls(
            id=message_id,
            current_version=1,
            versions=[
                MessageVersion(
                    version=1,
                    content=content,
                    edited_at=created_at,
                )
            ],
        )

    def edit(self, new_content: str, edited_at: int) -> None:
        if self.versions and edited_at < self.versions[-1].edited_at:
            raise ValueError("edited_at must be non-decreasing")

        next_version = self.current_version + 1
        self.versions.append(
            MessageVersion(
                version=next_version,
                content=new_content,
                edited_at=edited_at,
            )
        )
        self.current_version = next_version

    def latest(self) -> MessageVersion:
        return self.versions[self.current_version - 1]

    def as_of(self, timestamp: int) -> MessageVersion | None:
        left = 0
        right = len(self.versions) - 1
        answer = -1

        while left <= right:
            mid = (left + right) // 2

            if self.versions[mid].edited_at <= timestamp:
                answer = mid
                left = mid + 1
            else:
                right = mid - 1

        if answer == -1:
            return None
        return self.versions[answer]

def get_chat_messages(message_id: int, as_of: int | None = None) -> list[Message]:
    ...


def resolve_message(
    versioned_message: VersionedMessage,
    as_of: int | None,
) -> Message | None:
    version = (
        versioned_message.latest()
        if as_of is None
        else versioned_message.as_of(as_of)
    )

    if version is None:
        return None

    return Message(
        id=versioned_message.id,
        text=version.content,
    )

In a real system, you would usually store this as:

messages(message_id, current_version, author_id, created_at, ...)
message_versions(message_id, version, content, edited_at, editor_id, ...)

Why This Works

• Old content is never overwritten, so audit history and rollback remain possible.
• A stable message_id means the merge logic from Part 1 still deduplicates correctly.
• The latest version stays available in O(1) time.
• as_of(...) supports historical snapshots without changing the merge algorithm itself.

The important rule is still: deduplicate by message_id, not by version number.

Then decide whether get_chat_messages(...) should return:

• the latest version of each message
• the version visible at a snapshot time such as as_of

Complexity

• edit: O(1) append
• latest: O(1)
• as_of: O(log v) with binary search over v versions
• Extra space: O(v) per message to store version history

Source: darkinterview.com (real Reddit interview report, captured May 2026)

from typing import List

class SolutionBrute:
    def exist(self, board: List[List[str]], word: str) -> bool:
        rows, cols = len(board), len(board[0])
        visited = set()

        def dfs(r: int, c: int, idx: int) -> bool:
            # All characters matched — path found
            if idx == len(word):
                return True
            # Out of bounds
            if r < 0 or c < 0 or r >= rows or c >= cols:
                return False
            # Already used in this path
            if (r, c) in visited:
                return False
            # Character mismatch
            if board[r][c] != word[idx]:
                return False

            visited.add((r, c))
            found = (
                dfs(r + 1, c, idx + 1) or
                dfs(r - 1, c, idx + 1) or
                dfs(r, c + 1, idx + 1) or
                dfs(r, c - 1, idx + 1)
            )
            visited.discard((r, c))
            return found

        for r in range(rows):
            for c in range(cols):
                if dfs(r, c, 0):
                    return True
        return False


# --- smoke tests ---
if __name__ == "__main__":
    sol = SolutionBrute()
    board1 = [["A","B","C","E"],["S","F","C","S"],["A","D","E","E"]]
    assert sol.exist(board1, "ABCCED") is True
    assert sol.exist(board1, "SEE")    is True
    assert sol.exist(board1, "ABCB")   is False
    print("Brute-force: all tests passed")

from typing import List

class Solution:
    def exist(self, board: List[List[str]], word: str) -> bool:
        rows, cols = len(board), len(board[0])

        def dfs(r: int, c: int, idx: int) -> bool:
            # Base case: every character has been matched
            if idx == len(word):
                return True
            # Bounds check
            if r < 0 or c < 0 or r >= rows or c >= cols:
                return False
            # In-place visited check + character match combined:
            # '#' never equals word[idx], and wrong letter also fails here
            if board[r][c] != word[idx]:
                return False

            # Mark cell as visited by overwriting in-place
            tmp = board[r][c]
            board[r][c] = '#'

            # Explore all four neighbors
            found = (
                dfs(r + 1, c, idx + 1) or
                dfs(r - 1, c, idx + 1) or
                dfs(r, c + 1, idx + 1) or
                dfs(r, c - 1, idx + 1)
            )

            # Restore cell — critical for correctness of other search paths
            board[r][c] = tmp
            return found

        for r in range(rows):
            for c in range(cols):
                if dfs(r, c, 0):
                    return True
        return False


# --- comprehensive tests ---
if __name__ == "__main__":
    sol = Solution()

    board1 = [["A","B","C","E"],["S","F","C","S"],["A","D","E","E"]]
    assert sol.exist(board1, "ABCCED") is True,  "standard path"
    assert sol.exist(board1, "SEE")    is True,  "shorter path"
    assert sol.exist(board1, "ABCB")   is False, "backtrack required — must not revisit"

    # Single-cell edge cases
    assert sol.exist([["A"]], "A") is True
    assert sol.exist([["A"]], "B") is False

    # Word longer than board
    assert sol.exist([["A","B"],["C","D"]], "ABCD") is False  # can't make a valid path
    assert sol.exist([["A","B"],["C","D"]], "ABDC") is True

    # Repeated character trap
    board2 = [["A","A"]]
    assert sol.exist(board2, "AAA") is False  # only two A's
    assert sol.exist(board2, "AA")  is True

    print("Optimal: all tests passed")

from typing import List
from collections import Counter

class SolutionPruned:
    def exist(self, board: List[List[str]], word: str) -> bool:
        rows, cols = len(board), len(board[0])

        # --- Optimization A: frequency feasibility check ---
        board_count = Counter(c for row in board for c in row)
        word_count  = Counter(word)
        for ch, need in word_count.items():
            if board_count[ch] < need:
                return False   # not enough of this character on the board

        # --- Optimization B: search from the rarer end ---
        # Compare how often the first vs last character appears on the board.
        # If the last char is rarer, reversing means we start from fewer cells.
        if board_count[word[0]] > board_count[word[-1]]:
            word = word[::-1]   # reverse the target word

        # --- Core DFS with in-place marking (same as Approach 2) ---
        def dfs(r: int, c: int, idx: int) -> bool:
            if idx == len(word):
                return True
            if r < 0 or c < 0 or r >= rows or c >= cols:
                return False
            if board[r][c] != word[idx]:
                return False

            tmp = board[r][c]
            board[r][c] = '#'
            found = (
                dfs(r + 1, c, idx + 1) or
                dfs(r - 1, c, idx + 1) or
                dfs(r, c + 1, idx + 1) or
                dfs(r, c - 1, idx + 1)
            )
            board[r][c] = tmp
            return found

        for r in range(rows):
            for c in range(cols):
                if dfs(r, c, 0):
                    return True
        return False


# --- tests including pruning paths ---
if __name__ == "__main__":
    sol = SolutionPruned()

    board1 = [["A","B","C","E"],["S","F","C","S"],["A","D","E","E"]]
    assert sol.exist(board1, "ABCCED") is True
    assert sol.exist(board1, "SEE")    is True
    assert sol.exist(board1, "ABCB")   is False

    # Frequency prune: 'Z' does not appear on board at all
    assert sol.exist(board1, "ABCZ") is False

    # Reversed-word optimization: word ending in a rare character
    board3 = [["A","A","A","A"],
              ["A","A","A","A"],
              ["A","A","A","B"]]
    # "AAAAB" — 'B' is rare (1 cell); reverse to "BAAAA"
    assert sol.exist(board3, "AAAAB") is True
    assert sol.exist(board3, "AAAAAB") is False  # only one B

    print("Pruned: all tests passed")

Grid indices (row, col):
  (0,0)A  (0,1)B  (0,2)C  (0,3)E
  (1,0)S  (1,1)F  (1,2)C  (1,3)S
  (2,0)A  (2,1)D  (2,2)E  (2,3)E

DFS starts at (0,0)='A' matches word[0]='A'  → mark (0,0)='#'
  → try (1,0)='S' != 'B', fail
  → try (-1,0) out of bounds, fail
  → try (0,1)='B' matches word[1]='B'        → mark (0,1)='#'
      → try (0,2)='C' matches word[2]='C'    → mark (0,2)='#'
          → try (0,3)='E' != 'C', fail
          → try (1,2)='C' matches word[3]='C' → mark (1,2)='#'
              → try (2,2)='E' matches word[4]='E' → mark (2,2)='#'
                  → try (2,1)='D' matches word[5]='D' → mark (2,1)='#'
                      → dfs(_, _, 6): idx==len(word)==6 → return True ✓
              restore (2,2)='E'
          restore (1,2)='C'
      restore (0,2)='C'
  restore (0,1)='B'
restore (0,0)='A'
Result: True

Source: darkinterview.com (real Reddit interview report, captured May 2026)

from typing import List

class SolutionNaive:
    """
    Baseline: run Word Search I independently for every word.
    Time:  O(W * m * n * 4^L)  where W = #words, L = max word length
    Space: O(L) recursion stack per word
    Works, but TLEs on large inputs.
    """

    def findWords(self, board: List[List[str]], words: List[str]) -> List[str]:
        rows, cols = len(board), len(board[0])
        result = []

        def exists(word: str) -> bool:
            """Word Search I — can `word` be found on this board?"""
            def dfs(r: int, c: int, idx: int) -> bool:
                if idx == len(word):
                    return True
                if r < 0 or r >= rows or c < 0 or c >= cols:
                    return False
                if board[r][c] != word[idx]:
                    return False
                tmp, board[r][c] = board[r][c], "#"  # mark visited
                found = (
                    dfs(r + 1, c, idx + 1) or
                    dfs(r - 1, c, idx + 1) or
                    dfs(r, c + 1, idx + 1) or
                    dfs(r, c - 1, idx + 1)
                )
                board[r][c] = tmp  # restore
                return found

            return any(
                dfs(r, c, 0)
                for r in range(rows)
                for c in range(cols)
            )

        for word in words:
            if exists(word):
                result.append(word)
        return result


# --- quick sanity check ---
if __name__ == "__main__":
    board = [
        ["o","a","a","n"],
        ["e","t","a","e"],
        ["i","h","k","r"],
        ["i","f","l","v"],
    ]
    words = ["oath","pea","eat","rain"]
    print(SolutionNaive().findWords(board, words))  # ['eat', 'oath']

from typing import List

# ── Trie Node ──────────────────────────────────────────────────────────────────
class TrieNode:
    __slots__ = ("children", "word", "refs")

    def __init__(self):
        self.children: dict[str, "TrieNode"] = {}
        self.word: str | None = None   # non-None at end-of-word nodes
        self.refs: int = 0             # how many words pass through / end here


# ── Main Solution ──────────────────────────────────────────────────────────────
class Solution:
    """
    Trie + in-place DFS with dead-branch pruning.

    Time  (worst): O(m * n * 4^L)
          The 4^L factor comes from a single DFS path; the Trie prunes most of it.
          In practice, with a real word list the constant factor is dramatically
          smaller than the naive O(W * m * n * 4^L).

    Space: O(total_chars_in_all_words) for the Trie
           O(L) recursion stack (L = longest word)
    """

    # ── Phase 1: build Trie ────────────────────────────────────────────────────
    def _build_trie(self, words: List[str]) -> TrieNode:
        root = TrieNode()
        for word in words:
            node = root
            node.refs += 1
            for ch in word:
                if ch not in node.children:
                    node.children[ch] = TrieNode()
                node = node.children[ch]
                node.refs += 1
            node.word = word
        return root

    # ── Phase 2: prune a branch that is no longer needed ──────────────────────
    def _prune(self, root: TrieNode, word: str) -> None:
        """
        Walk `word` down from root, decrementing `refs`.
        Delete a child pointer whenever refs drops to 0.
        This shrinks the Trie as words are collected, so future DFS calls
        skip dead branches immediately.
        """
        node = root
        node.refs -= 1
        path: list[tuple[TrieNode, str]] = []  # (parent, char)
        for ch in word:
            path.append((node, ch))
            node = node.children[ch]
            node.refs -= 1
        # walk back and delete zero-ref nodes
        for parent, ch in reversed(path):
            if parent.children[ch].refs == 0:
                del parent.children[ch]
            else:
                break  # ancestors still have live words, stop

    # ── Phase 3: DFS ──────────────────────────────────────────────────────────
    def findWords(self, board: List[List[str]], words: List[str]) -> List[str]:
        root = self._build_trie(words)
        rows, cols = len(board), len(board[0])
        result: list[str] = []

        def dfs(r: int, c: int, node: TrieNode) -> None:
            # ① Out-of-bounds check
            if r < 0 or r >= rows or c < 0 or c >= cols:
                return
            ch = board[r][c]
            # ② Cell was already visited on this path, or no Trie edge
            if ch == "#" or ch not in node.children:
                return
            nxt = node.children[ch]
            # ③ Collect word if we've reached an end-of-word node
            if nxt.word:
                result.append(nxt.word)
                nxt.word = None        # prevent duplicate collection
                self._prune(root, result[-1])  # remove dead branch
                # Note: after pruning, nxt may no longer be reachable via
                # node.children[ch]. We must NOT use nxt below if pruned.
                # Re-check whether the edge still exists:
                if ch not in node.children:
                    return
                nxt = node.children[ch]  # re-fetch (same object, not deleted)

            # ④ Mark visited in-place (O(1), avoids a separate visited matrix)
            board[r][c] = "#"
            dfs(r + 1, c, nxt)
            dfs(r - 1, c, nxt)
            dfs(r, c + 1, nxt)
            dfs(r, c - 1, nxt)
            board[r][c] = ch   # ⑤ restore

        for r in range(rows):
            for c in range(cols):
                if board[r][c] in root.children:  # fast early-exit
                    dfs(r, c, root)

        return result


# ── Tests ──────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    sol = Solution()

    # Example 1 (from LC)
    board1 = [
        ["o","a","a","n"],
        ["e","t","a","e"],
        ["i","h","k","r"],
        ["i","f","l","v"],
    ]
    words1 = ["oath","pea","eat","rain"]
    out1 = sorted(sol.findWords(board1, words1))
    assert out1 == ["eat","oath"], out1

    # Example 2 (LC)
    board2 = [["a","b"],["c","d"]]
    words2 = ["abcb"]
    out2 = sol.findWords(board2, words2)
    assert out2 == [], out2  # 'b' cannot be visited twice

    # Single-cell board
    board3 = [["a"]]
    out3 = sol.findWords(board3, ["a"])
    assert out3 == ["a"], out3

    # All same letter — stress test for pruning
    board4 = [["a"]*4 for _ in range(4)]
    words4 = ["aaa","aaaa","aaaaa"]
    out4 = sorted(sol.findWords(board4, words4))
    # "aaa", "aaaa", "aaaaa" can all be formed on a 4x4 board of 'a'
    assert set(out4) == {"aaa","aaaa","aaaaa"}, out4

    print("All tests passed.")

from typing import List

class TrieNode:
    __slots__ = ("children", "word")

    def __init__(self):
        self.children: dict[str, "TrieNode"] = {}
        self.word: str | None = None


class SolutionIterative:
    """
    Same Trie build, but DFS is iterative to avoid hitting Python's
    recursion limit on large boards.

    Time / Space: identical to Approach 2.
    """

    def _build_trie(self, words: List[str]) -> TrieNode:
        root = TrieNode()
        for word in words:
            node = root
            for ch in word:
                if ch not in node.children:
                    node.children[ch] = TrieNode()
                node = node.children[ch]
            node.word = word
        return root

    def findWords(self, board: List[List[str]], words: List[str]) -> List[str]:
        root = self._build_trie(words)
        rows, cols = len(board), len(board[0])
        result: list[str] = []
        found_set: set[str] = set()

        for start_r in range(rows):
            for start_c in range(cols):
                ch = board[start_r][start_c]
                if ch not in root.children:
                    continue

                # Stack item: (r, c, trie_node, restore_list)
                # restore_list accumulates (r, c, original_char) to undo on pop.
                # We use a sentinel (None, None, None) to trigger restoration.
                stack: list = [(start_r, start_c, root, [])]

                while stack:
                    item = stack[-1]

                    if item is None:
                        # Sentinel — restore the cell above this frame
                        stack.pop()
                        r_restore, c_restore, orig_ch = stack.pop()
                        board[r_restore][c_restore] = orig_ch
                        continue

                    r, c, node = item
                    stack.pop()

                    cur_ch = board[r][c]
                    if cur_ch == "#" or cur_ch not in node.children:
                        continue

                    nxt = node.children[cur_ch]

                    if nxt.word and nxt.word not in found_set:
                        found_set.add(nxt.word)
                        result.append(nxt.word)
                        nxt.word = None

                    # Mark visited
                    board[r][c] = "#"
                    # Push restore sentinel AFTER the children so it fires last
                    stack.append((r, c, cur_ch))   # restore data
                    stack.append(None)              # sentinel

                    for dr, dc in ((1,0),(-1,0),(0,1),(0,-1)):
                        nr, nc = r + dr, c + dc
                        if 0 <= nr < rows and 0 <= nc < cols:
                            stack.append((nr, nc, nxt))

        return result


if __name__ == "__main__":
    sol = SolutionIterative()
    board = [
        ["o","a","a","n"],
        ["e","t","a","e"],
        ["i","h","k","r"],
        ["i","f","l","v"],
    ]
    words = ["oath","pea","eat","rain"]
    print(sorted(sol.findWords(board, words)))  # ['eat', 'oath']

Source: darkinterview.com (real Reddit interview report, captured May 2026)

Problem Overview

You are given:

• a start word
• a target word
• a dictionary of allowed words

Determine whether the words can be connected by a valid transformation sequence.

Each step in the sequence must move from the current word to another word of the same length. The interviewer usually starts with the standard one-character version, then adds a follow-up where a move may change one or two characters.

This is essentially a Word Ladder style graph problem:

• each word is a node
• an edge exists when two words differ by an allowed number of character positions
• the task is to search for a path from start to target

Assume:

• all usable words have the same length
• start and target are lowercase strings
• the dictionary may contain duplicates; treat it as a set
• start does not need to be in the dictionary
• target should still be considered a valid destination even if it is not already in the dictionary
• words should not be revisited once already explored

Part 1: Reachability With One-Character Transforms

Problem Statement

Implement:

def has_path_one_edit(start: str, target: str, words: list[str]) -> bool:
    ...

Return True if there exists a sequence from start to target such that:

• every step changes exactly one character
• every intermediate word is in the dictionary
• target may be used as the final word even if it is not in the dictionary

Example

start = "hit"
target = "cog"
words = ["hot", "dot", "dog", "lot", "log", "cog"]

has_path_one_edit(start, target, words)  # True

One valid path is:

hit -> hot -> dot -> dog -> cog

Part 1 Solution

For plain reachability, either DFS or BFS works. BFS is a good default because it extends naturally to the shortest-path follow-up later.

from collections import deque


def hamming_distance(a: str, b: str) -> int:
    return sum(char_a != char_b for char_a, char_b in zip(a, b))


def has_path_one_edit(start: str, target: str, words: list[str]) -> bool:
    if len(start) != len(target):
        return False
    if start == target:
        return True

    remaining = {word for word in words if len(word) == len(start)}
    remaining.add(target)

    queue = deque([start])
    visited = {start}

    while queue:
        word = queue.popleft()

        for candidate in list(remaining):
            if candidate in visited:
                continue

            if hamming_distance(word, candidate) == 1:
                if candidate == target:
                    return True

                visited.add(candidate)
                queue.append(candidate)

        remaining -= visited

    return False

Complexity

Let n be the number of candidate words and m be the word length
Time: O(n^2 * m) in the straightforward implementation
Space: O(n)

Part 2: Reachability With One- Or Two-Character Transforms

Problem Statement

Now extend the rule:

• a valid move may change either exactly one character or exactly two characters

Implement:

def has_path_one_or_two_edits(start: str, target: str, words: list[str]) -> bool:
    ...

Example

start = "code"
target = "math"
words = ["coda", "cada", "mata"]

has_path_one_or_two_edits(start, target, words)  # True

One valid path is:

code -> coda -> cada -> mata -> math

The key jump is:

cada -> mata

which changes two character positions.

Part 2 Solution

The search logic stays the same. The main change is the adjacency rule.

One reason this follow-up is trickier than standard Word Ladder is that the classic single-wildcard bucket trick is built for one-character neighbors. Once two-character jumps are allowed, the simplest interview-safe solution is often to keep the graph implicit and compare candidate words directly.

from collections import deque


def hamming_distance(a: str, b: str) -> int:
    return sum(char_a != char_b for char_a, char_b in zip(a, b))


def has_path_one_or_two_edits(start: str, target: str, words: list[str]) -> bool:
    if len(start) != len(target):
        return False
    if start == target:
        return True

    remaining = {word for word in words if len(word) == len(start)}
    remaining.add(target)

    queue = deque([start])
    visited = {start}

    while queue:
        word = queue.popleft()

        for candidate in list(remaining):
            if candidate in visited:
                continue

            distance = hamming_distance(word, candidate)
            if distance == 1 or distance == 2:
                if candidate == target:
                    return True

                visited.add(candidate)
                queue.append(candidate)

        remaining -= visited

    return False

Complexity

Time: O(n^2 * m)
Space: O(n)

The asymptotic complexity is unchanged from Part 1. Only the edge predicate changed.

Part 3: Return An Actual Path

Problem Statement

Now return one shortest valid path instead of only a boolean.

Implement:

def shortest_path_one_or_two_edits(
    start: str,
    target: str,
    words: list[str],
) -> list[str]:
    ...

Return:

• a shortest valid transformation sequence from start to target, inclusive
• [] if no path exists

Example

start = "code"
target = "math"
words = ["coda", "cada", "mata"]

shortest_path_one_or_two_edits(start, target, words)
# ["code", "coda", "cada", "mata", "math"]

Part 3 Solution

Once the interviewer asks for a shortest path, BFS is the right tool. Keep a parent map so you can reconstruct the path after reaching target.

from collections import deque


def hamming_distance(a: str, b: str) -> int:
    return sum(char_a != char_b for char_a, char_b in zip(a, b))


def shortest_path_one_or_two_edits(
    start: str,
    target: str,
    words: list[str],
) -> list[str]:
    if len(start) != len(target):
        return []
    if start == target:
        return [start]

    remaining = {word for word in words if len(word) == len(start)}
    remaining.add(target)

    queue = deque([start])
    parent: dict[str, str | None] = {start: None}

    while queue:
        word = queue.popleft()

        if word == target:
            break

        next_words: list[str] = []
        for candidate in remaining:
            distance = hamming_distance(word, candidate)
            if distance == 1 or distance == 2:
                next_words.append(candidate)

        for candidate in next_words:
            if candidate in parent:
                continue

            parent[candidate] = word
            queue.append(candidate)

        remaining -= set(next_words)

    if target not in parent:
        return []

    path: list[str] = []
    current: str | None = target
    while current is not None:
        path.append(current)
        current = parent[current]

    path.reverse()
    return path

Complexity

Time: O(n^2 * m)
Space: O(n)

Discussion Points

Interviewers often use this question to check whether the candidate can:

• model the problem as a graph instead of getting stuck on string manipulation
• explain when DFS is enough versus when BFS is the better choice
• handle the two-character follow-up without overcomplicating the solution too early
• reason about neighbor-generation tradeoffs

Good follow-up discussion:

• DFS is sufficient if the task is only "does any path exist?"
• BFS is better if the task changes to "find the shortest path"
• for large dictionaries, precomputed neighbor indexes can help Part 1
• Part 2 is less friendly to the standard Word Ladder wildcard optimization, so a direct Hamming-distance scan is often the cleanest baseline answer

Source: darkinterview.com (real Reddit interview report, captured May 2026)

Problem Overview

We accidentally dropped the database that stored the current billing state for advertisers. The old transaction logs still exist, so the task is to replay those transactions and rebuild one BillingStatus object per user.

This interview is usually framed as an object-oriented coding problem. The core idea is to model a single account as a BillingStatus class that can ingest transactions over time, then build a dictionary like:

{
    user_id: BillingStatus(...),
    user_id_2: BillingStatus(...),
}

The interviewer typically adds three layers:

• Basic additive aggregation
• Overwrite transactions
• undo_last and redo_last

Assume:

• replay order is increasing transaction_timestamp
• if two transactions have the same timestamp, break ties by transaction_id for deterministic replay
• monetary_columns is known in advance
• missing monetary fields mean "no change" for that column
• undo_last and redo_last only affect the history of the same user
• overwrite, undo_last, and redo_last are control fields, not monetary columns
• if a transaction uses undo_last or redo_last, treat it as a command row and ignore any monetary fields on that same row
• redo_last follows standard stack semantics: it reapplies the most recently undone regular transaction
• each transaction uses at most one of overwrite, undo_last, or redo_last as its behavioral option, except that plain regular transactions may omit all of them

Part 1: Rebuild Billing Statuses

Problem Statement

Implement a BillingStatus class with two starting monetary columns:

ad_delivery_pennies = 0
payment_pennies = 0

Each transaction may contain one or more monetary columns. When ingesting a transaction, add the transaction values into the current billing status.

Then implement a function that replays a collection of transactions and returns one BillingStatus per user.

Example

monetary_columns = ("ad_delivery_pennies", "payment_pennies")

transactions = {
    "ff8bc1c2-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "ad_delivery_pennies": 1000,
        "transaction_timestamp": 1500000001,
    },
    "ff8bc2e4-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "ad_delivery_pennies": 1000,
        "transaction_timestamp": 1500000002,
    },
    "ff8bc4ec-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "payment_pennies": 500,
        "transaction_timestamp": 1500000003,
    },
    "fv24z4ec-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "ad_delivery_pennies": 1000,
        "payment_pennies": 500,
        "transaction_timestamp": 1500000004,
    },
}

# Expected result:
{
    1: BillingStatus(
        ad_delivery_pennies=3000,
        payment_pennies=1000,
    )
}

Part 1 Solution

Use a per-user object and replay the log in timestamp order.

class BillingStatus:
    def __init__(self, monetary_columns: tuple[str, ...]):
        self.monetary_columns = tuple(monetary_columns)
        self.amounts = {column: 0 for column in self.monetary_columns}

    def ingest(self, transaction_id: str, transaction: dict) -> None:
        for column in self.monetary_columns:
            self.amounts[column] += transaction.get(column, 0)

    def as_dict(self) -> dict[str, int]:
        return dict(self.amounts)


def rebuild_billing_statuses(
    transactions: dict[str, dict],
    monetary_columns: tuple[str, ...],
) -> dict[int, BillingStatus]:
    statuses: dict[int, BillingStatus] = {}

    ordered_transactions = sorted(
        transactions.items(),
        key=lambda item: (item[1]["transaction_timestamp"], item[0]),
    )

    for transaction_id, transaction in ordered_transactions:
        user_id = transaction["user_id"]
        status = statuses.setdefault(user_id, BillingStatus(monetary_columns))
        status.ingest(transaction_id, transaction)

    return statuses

Complexity

Sorting: O(n log n)
Replay: O(n * c), where c is the number of monetary columns
Space: O(u * c), where u is the number of users

Part 2: Add Overwrite Transactions

Problem Statement

Now support a control flag:

"overwrite": True

If overwrite is True, any monetary column present in that transaction should replace the current value for that column instead of being added to it.

Important detail:

• overwrite only applies to columns present in the transaction
• columns not mentioned in the transaction should remain unchanged

Example

monetary_columns = ("ad_delivery_pennies", "payment_pennies")

transactions = {
    "ff8ba98a-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "ad_delivery_pennies": 1000,
        "transaction_timestamp": 1500000001,
        "overwrite": False,
    },
    "ff8bad4a-8d45-11e9-bc42-526af7764f64": {
        "user_id": 2,
        "ad_delivery_pennies": 1000,
        "transaction_timestamp": 1500000004,
    },
    "ff8baea8-8d45-11e9-bc42-526af7764f64": {
        "user_id": 2,
        "payment_pennies": 600,
        "transaction_timestamp": 1500000007,
        "overwrite": False,
    },
    "ff8bb4ac-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "ad_delivery_pennies": 1000,
        "transaction_timestamp": 1500000002,
        "overwrite": False,
    },
    "ff8bb600-8d45-11e9-bc42-526af7764f64": {
        "user_id": 2,
        "ad_delivery_pennies": 1000,
        "payment_pennies": 500,
        "transaction_timestamp": 1500000003,
        "overwrite": False,
    },
    "ff8bb89e-8d45-11e9-bc42-526af7764f64": {
        "user_id": 2,
        "payment_pennies": 2000,
        "transaction_timestamp": 1500000005,
        "overwrite": True,
    },
    "ff8bb9c0-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "payment_pennies": 500,
        "transaction_timestamp": 1500000003,
        "overwrite": False,
    },
    "ff8bbf74-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "ad_delivery_pennies": 1000,
        "payment_pennies": 500,
        "transaction_timestamp": 1500000004,
        "overwrite": True,
    },
    "ff8bc0a0-8d45-11e9-bc42-526af7764f64": {
        "user_id": 2,
        "ad_delivery_pennies": 1000,
        "transaction_timestamp": 1500000001,
    },
    "ff8bc1c2-8d45-11e9-bc42-526af7764f64": {
        "user_id": 2,
        "ad_delivery_pennies": 1000,
        "transaction_timestamp": 1500000002,
    },
    "ff923488-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "payment_pennies": 100,
        "transaction_timestamp": 1500000013,
    },
}

# Expected result:
{
    1: BillingStatus(ad_delivery_pennies=1000, payment_pennies=600),
    2: BillingStatus(ad_delivery_pennies=4000, payment_pennies=2600),
}

Part 2 Solution

The replay loop does not change. Only BillingStatus.ingest(...) changes:

class BillingStatus:
    def __init__(self, monetary_columns: tuple[str, ...]):
        self.monetary_columns = tuple(monetary_columns)
        self.amounts = {column: 0 for column in self.monetary_columns}

    def ingest(self, transaction_id: str, transaction: dict) -> None:
        overwrite = transaction.get("overwrite", False)

        for column in self.monetary_columns:
            if column not in transaction:
                continue

            if overwrite:
                self.amounts[column] = transaction[column]
            else:
                self.amounts[column] += transaction[column]

    def as_dict(self) -> dict[str, int]:
        return dict(self.amounts)

Why This Matches The Example

For user 1:

• ad += 1000
• ad += 1000
• payment += 500
• overwrite both columns to ad = 1000, payment = 500
• payment += 100

Final state:

{
    "ad_delivery_pennies": 1000,
    "payment_pennies": 600,
}

For user 2, the overwrite only touches payment_pennies, so ad_delivery_pennies keeps its accumulated value.

Part 3: Add undo_last And redo_last

Problem Statement

Now add two more control flags:

"undo_last": True
"redo_last": True

Rules:

• undo_last=True undoes the most recent regular transaction for the same user
• redo_last=True reapplies the most recently undone regular transaction for the same user
• if there is nothing to undo, discard the operation
• if there is nothing to redo, discard the operation
• a regular transaction is:
  • a transaction with no control flags, or
  • a transaction that only uses overwrite
• assume a transaction never sets both undo_last and redo_last
• undo_last and redo_last transactions are commands, not regular transactions
• if a new regular transaction is applied after an undo, the redo history should be cleared
• if a command row also contains monetary columns, ignore those monetary columns and only execute the command

To remove ambiguity, assume standard editor-style undo/redo semantics:

• undo pops from the applied-history stack
• redo pops from the undone-history stack
• any newly applied regular transaction invalidates redo history

Example

monetary_columns = ("ad_delivery_pennies", "payment_pennies")

transactions = {
    "ff8bc1c2-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "ad_delivery_pennies": 1000,
        "transaction_timestamp": 1500000001,
    },
    "ff8bc2e4-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "undo_last": True,
        "transaction_timestamp": 1500000002,
    },
    "ff8bc4ec-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "payment_pennies": 500,
        "transaction_timestamp": 1500000003,
    },
    "fv24z4ec-8d45-11e9-bc42-526af7764f64": {
        "user_id": 1,
        "ad_delivery_pennies": 1000,
        "payment_pennies": 500,
        "transaction_timestamp": 1500000004,
    },
}

# Expected result:
{
    1: BillingStatus(ad_delivery_pennies=1000, payment_pennies=1000),
}

Part 3 Solution

Store enough history to reverse and reapply the effect of each regular transaction.

from dataclasses import dataclass


@dataclass
class ReplayRecord:
    columns: tuple[str, ...]
    before: dict[str, int]
    after: dict[str, int]


class BillingStatus:
    def __init__(self, monetary_columns: tuple[str, ...]):
        self.monetary_columns = tuple(monetary_columns)
        self.amounts = {column: 0 for column in self.monetary_columns}
        self._applied: list[ReplayRecord] = []
        self._undone: list[ReplayRecord] = []

    def ingest(self, transaction_id: str, transaction: dict) -> None:
        if transaction.get("undo_last"):
            self._undo_last()
            return

        if transaction.get("redo_last"):
            self._redo_last()
            return

        touched_columns = tuple(
            column
            for column in self.monetary_columns
            if column in transaction
        )

        if not touched_columns:
            return

        overwrite = transaction.get("overwrite", False)
        before = {column: self.amounts[column] for column in touched_columns}

        for column in touched_columns:
            value = transaction[column]
            if overwrite:
                self.amounts[column] = value
            else:
                self.amounts[column] += value

        after = {column: self.amounts[column] for column in touched_columns}
        self._applied.append(
            ReplayRecord(
                columns=touched_columns,
                before=before,
                after=after,
            )
        )
        self._undone.clear()

    def _undo_last(self) -> None:
        if not self._applied:
            return

        record = self._applied.pop()
        for column in record.columns:
            self.amounts[column] = record.before[column]
        self._undone.append(record)

    def _redo_last(self) -> None:
        if not self._undone:
            return

        record = self._undone.pop()
        for column in record.columns:
            self.amounts[column] = record.after[column]
        self._applied.append(record)

    def as_dict(self) -> dict[str, int]:
        return dict(self.amounts)


def rebuild_billing_statuses(
    transactions: dict[str, dict],
    monetary_columns: tuple[str, ...],
) -> dict[int, BillingStatus]:
    statuses: dict[int, BillingStatus] = {}

    ordered_transactions = sorted(
        transactions.items(),
        key=lambda item: (item[1]["transaction_timestamp"], item[0]),
    )

    for transaction_id, transaction in ordered_transactions:
        user_id = transaction["user_id"]
        status = statuses.setdefault(user_id, BillingStatus(monetary_columns))
        status.ingest(transaction_id, transaction)

    return statuses

Why This Design Works

• BillingStatus owns the mutable state for one user
• each regular transaction stores the exact before and after values for the touched columns
• undo_last restores before
• redo_last restores after
• overwrite transactions work naturally because the history captures absolute values, not only deltas

This is usually the main OOD discussion point: do you store only deltas, or do you store reversible command history? Once overwrite exists, storing before and after snapshots for touched columns keeps the implementation simple and correct.

Complexity

Let c be the number of monetary columns.

• regular ingest: O(c)
• undo_last: O(k)
• redo_last: O(k)

where k is the number of columns touched by that transaction, and k <= c.

Overall replay remains:

• Sorting: O(n log n)
• Replay: O(n * c)
• Space: O(u * c + h), where h is the per-user history retained for undo/redo

Common Follow-Ups

Interviewers may push on a few practical extensions:

Idempotency

• What if the logs contain duplicate transaction IDs?
• One answer is to keep a per-user or global seen_transaction_ids set.

Streaming ingestion

• The same BillingStatus.ingest(...) API works for both historical replay and live traffic.

Checkpointing

• If the logs are huge, periodically snapshot BillingStatus and only replay newer transactions after the latest checkpoint.

Auditability

• In production you would likely store the command history, not just the final balances, so that later investigations can explain why a balance has its current value.

Source: darkinterview.com (real Reddit interview report, captured May 2026)

class SolutionBrute:
    def shortestPalindrome(self, s: str) -> str:
        # Edge case: empty or single character is already a palindrome
        if len(s) <= 1:
            return s

        def is_palindrome(t: str) -> bool:
            return t == t[::-1]

        # Try the longest possible palindromic prefix first (greedy from the top).
        # s[:n], s[:n-1], ..., s[:1] — stop at the first palindrome.
        n = len(s)
        for k in range(n, 0, -1):
            if is_palindrome(s[:k]):
                # Characters not in the palindromic prefix must be prepended in reverse
                suffix = s[k:]          # the "remainder" that is NOT part of the prefix
                return suffix[::-1] + s # prepend its reverse

        # Fallback: prepend the reverse of the entire string minus the first char
        # (single char is always a palindrome, so k=1 always qualifies above)
        return s[::-1] + s   # unreachable in practice, but keeps the function total


# --- tests ---
def _test_brute():
    sol = SolutionBrute()
    assert sol.shortestPalindrome("aacecaaa") == "aaacecaaa"
    assert sol.shortestPalindrome("abcd")     == "dcbabcd"
    assert sol.shortestPalindrome("")         == ""
    assert sol.shortestPalindrome("a")        == "a"
    assert sol.shortestPalindrome("aa")       == "aa"
    assert sol.shortestPalindrome("aba")      == "aba"
    assert sol.shortestPalindrome("abba")     == "abba"
    assert sol.shortestPalindrome("race")     == "ecarace"
    print("brute force: all tests passed")

_test_brute()

class Solution:
    def shortestPalindrome(self, s: str) -> str:
        if not s:
            return s

        rev = s[::-1]

        # Build the combined string with a separator that cannot appear in the
        # input alphabet.  The '#' sentinel is sufficient for ASCII lowercase.
        combined = s + '#' + rev   # length = 2n + 1

        # ---------------------------------------------------------------
        # Compute the KMP "failure function" (a.k.a. prefix function / lps).
        # lps[i] = length of the longest proper prefix of combined[:i+1]
        #          that is also a suffix of combined[:i+1].
        # ---------------------------------------------------------------
        n = len(combined)
        lps = [0] * n

        # We process indices 1 .. n-1 (lps[0] is always 0 by definition).
        # 'length' tracks the length of the previous longest prefix-suffix.
        length = 0
        i = 1
        while i < n:
            if combined[i] == combined[length]:
                # Characters match: extend the current prefix-suffix.
                length += 1
                lps[i] = length
                i += 1
            else:
                if length != 0:
                    # Fall back using the already-computed table.
                    # We do NOT increment i here — we just try a shorter prefix.
                    length = lps[length - 1]
                else:
                    # No prefix-suffix possible; move on.
                    lps[i] = 0
                    i += 1

        # lps[-1] is the length of the longest palindromic prefix of s.
        pal_prefix_len = lps[-1]

        # The characters of s beyond the palindromic prefix must be prepended
        # in reverse to complete the palindrome.
        suffix = s[pal_prefix_len:]       # the "uncovered" tail of s
        return suffix[::-1] + s           # prepend reversed tail


# ---------------------------------------------------------------
# Worked trace for s = "aacecaaa"
# ---------------------------------------------------------------
# rev           = "aaacecaa"
# combined      = "aacecaaa#aaacecaa"   (length 17)
#
# KMP prefix function step-by-step (only key transitions shown):
#  i  char   length   lps[i]
#  0   a       —       0     (by definition)
#  1   a       0 → 1   1
#  2   c       0       0
#  3   e       0       0
#  4   c       0       0
#  5   a       0 → 1   1
#  6   a       1 → 2   2
#  7   a       2 → 3   3
#  8   #       3 → … → 0   0   (# never matches 'a', so fall all the way to 0)
#  9   a       0 → 1   1
# 10   a       1 → 2   2
# 11   a       2 → 3   3
# 12   c       3 → (fallback to lps[2]=0) → 0   0
#              — wait, 'c' != combined[3]='e', and lps[2]=0, so 0
#              Actually combined[3]='e', combined[12]='c' → no match, length=lps[2]=0
#              Then combined[0]='a' vs 'c' → no match, lps[12]=0
# 13   e       0 → 0 (combined[0]='a' != 'e') → 0
# 14   c       0 → 0 (combined[0]='a' != 'c') → 0
# 15   a       0 → 1   1
# 16   a       1 → 2   2  ← THIS is lps[-1]
#              But wait — combined[1]='a', combined[16]='a' ... let's re-examine.
#              Longest palindromic prefix of "aacecaaa" is "aacecaa" (length 7)?
#              No — "aacecaaa" reversed is "aaacecaa". Prefix "aacecaa" (k=7)?
#              "aacecaa" reversed = "aacecaa" → yes, it's a palindrome!  length=7
#              BUT we said lps[-1]=7? Let's recount: lps[-1] gives the match length.
#              "aacecaaa"[:7] = "aacecaa" is indeed a palindrome.
#              suffix = s[7:] = "a" → reversed = "a" → answer = "a" + "aacecaaa" = "aaacecaaa" ✓
#
# ---------------------------------------------------------------
# Worked trace for s = "abcd"
# ---------------------------------------------------------------
# rev      = "dcba"
# combined = "abcd#dcba"  (length 9)
# lps[-1]  = 1  (only "a" is a palindromic prefix)
# suffix   = "bcd" → reversed = "dcb"
# answer   = "dcb" + "abcd" = "dcbabcd"  ✓


# --- tests ---
def _test_kmp():
    sol = Solution()
    assert sol.shortestPalindrome("aacecaaa") == "aaacecaaa", sol.shortestPalindrome("aacecaaa")
    assert sol.shortestPalindrome("abcd")     == "dcbabcd"
    assert sol.shortestPalindrome("")         == ""
    assert sol.shortestPalindrome("a")        == "a"
    assert sol.shortestPalindrome("aa")       == "aa"
    assert sol.shortestPalindrome("aba")      == "aba"
    assert sol.shortestPalindrome("abba")     == "abba"
    assert sol.shortestPalindrome("race")     == "ecarace"
    # adversarial: all same chars + one different
    assert sol.shortestPalindrome("aaab")     == "baaab"
    # entire string already a palindrome
    assert sol.shortestPalindrome("racecar")  == "racecar"
    print("KMP solution: all tests passed")

_test_kmp()

class SolutionRollingHash:
    def shortestPalindrome(self, s: str) -> str:
        if not s:
            return s

        # Two independent (base, mod) pairs to reduce collision probability.
        BASE1, MOD1 = 131, (1 << 61) - 1   # Mersenne prime — fast mod
        BASE2, MOD2 = 137, (1 << 31) - 1

        n = len(s)

        # Precompute forward hash: hash(s[:k]) for all k in [1..n].
        # fwd[k] = hash of s[0..k-1]
        fwd1 = [0] * (n + 1)
        fwd2 = [0] * (n + 1)
        for i, ch in enumerate(s):
            fwd1[i + 1] = (fwd1[i] * BASE1 + ord(ch)) % MOD1
            fwd2[i + 1] = (fwd2[i] * BASE2 + ord(ch)) % MOD2

        # Precompute reverse hash: hash(s[k-1..0]) for all k.
        # rev[k] = hash of s[:k] read right-to-left  (i.e., hash of s[:k][::-1])
        pow1 = [1] * (n + 1)
        pow2 = [1] * (n + 1)
        for i in range(1, n + 1):
            pow1[i] = pow1[i - 1] * BASE1 % MOD1
            pow2[i] = pow2[i - 1] * BASE2 % MOD2

        # rev_hash[k] = polynomial hash of the reversed prefix s[:k]
        # Build it incrementally: append s[k-1] as the new *leading* character.
        # rev_hash[k] = ord(s[k-1]) * BASE^(k-1)  +  rev_hash[k-1]
        rev1 = [0] * (n + 1)
        rev2 = [0] * (n + 1)
        for i in range(1, n + 1):
            rev1[i] = (rev1[i - 1] + ord(s[i - 1]) * pow1[i - 1]) % MOD1
            rev2[i] = (rev2[i - 1] + ord(s[i - 1]) * pow2[i - 1]) % MOD2

        # But the forward hash of s[:k] uses a *different* polynomial convention:
        # fwd[k] = s[0]*B^(k-1) + s[1]*B^(k-2) + ... + s[k-1]*B^0
        # rev[k] should match fwd[k] when the prefix is a palindrome.
        # For them to be comparable we need the same polynomial convention.
        # Recompute rev_hash using the same left-to-right convention on the reversed string.
        rev_s = s[::-1]
        rfwd1 = [0] * (n + 1)
        rfwd2 = [0] * (n + 1)
        for i, ch in enumerate(rev_s):
            rfwd1[i + 1] = (rfwd1[i] * BASE1 + ord(ch)) % MOD1
            rfwd2[i + 1] = (rfwd2[i] * BASE2 + ord(ch)) % MOD2

        # Helper: extract hash of rev_s[n-k .. n-1]  (= reverse of s[:k])
        # using suffix extraction from rfwd.
        # rfwd covers the entire reversed string; we want the LAST k characters of rev_s,
        # which correspond to the first k characters of s reversed.
        # rfwd[n] covers all of rev_s.  The last k chars of rev_s are rev_s[n-k..n-1].
        # hash(rev_s[n-k..n-1]) = rfwd[n] - rfwd[n-k] * BASE^k  (standard Rabin-Karp)
        def rev_hash_of_prefix(k):
            h1 = (rfwd1[n] - rfwd1[n - k] * pow1[k]) % MOD1
            h2 = (rfwd2[n] - rfwd2[n - k] * pow2[k]) % MOD2
            return h1, h2

        # Scan from the longest possible palindromic prefix down.
        # We want the largest k such that hash(s[:k]) == hash(reverse(s[:k])).
        best_k = 1   # single character is always a palindrome
        for k in range(n, 0, -1):
            # hash of s[:k]
            h_fwd = (fwd1[k], fwd2[k])
            # hash of reverse(s[:k])
            h_rev = rev_hash_of_prefix(k)
            if h_fwd == h_rev:
                best_k = k
                break

        suffix = s[best_k:]
        return suffix[::-1] + s


# --- tests ---
def _test_hash():
    sol = SolutionRollingHash()
    assert sol.shortestPalindrome("aacecaaa") == "aaacecaaa"
    assert sol.shortestPalindrome("abcd")     == "dcbabcd"
    assert sol.shortestPalindrome("")         == ""
    assert sol.shortestPalindrome("a")        == "a"
    assert sol.shortestPalindrome("aba")      == "aba"
    assert sol.shortestPalindrome("race")     == "ecarace"
    assert sol.shortestPalindrome("aaab")     == "baaab"
    assert sol.shortestPalindrome("racecar")  == "racecar"
    print("rolling hash solution: all tests passed")

_test_hash()

Source: darkinterview.com (real Reddit interview report, captured May 2026)

from typing import Optional

class ListNode:
    def __init__(self, val: int = 0, next: 'Optional[ListNode]' = None):
        self.val = val
        self.next = next

class SolutionBrute:
    def oddEvenList(self, head: Optional[ListNode]) -> Optional[ListNode]:
        # O(n) time, O(n) space — NOT acceptable per the problem constraint.
        # Stated here to anchor the discussion.
        if not head:
            return head

        odds, evens = [], []
        node, idx = head, 1
        while node:
            if idx % 2 == 1:
                odds.append(node.val)
            else:
                evens.append(node.val)
            node = node.next
            idx += 1

        # Overwrite values in the original list
        node = head
        for v in odds + evens:
            node.val = v
            node = node.next

        return head

# ---- helpers for testing ----
def make_list(vals):
    dummy = ListNode(0)
    cur = dummy
    for v in vals:
        cur.next = ListNode(v)
        cur = cur.next
    return dummy.next

def list_vals(head):
    out = []
    while head:
        out.append(head.val)
        head = head.next
    return out

# smoke test
assert list_vals(SolutionBrute().oddEvenList(make_list([1,2,3,4,5]))) == [1,3,5,2,4]
assert list_vals(SolutionBrute().oddEvenList(make_list([2,1,3,5,6,4,7]))) == [2,3,6,7,1,5,4]
assert list_vals(SolutionBrute().oddEvenList(make_list([]))) == []
assert list_vals(SolutionBrute().oddEvenList(make_list([1]))) == [1]
assert list_vals(SolutionBrute().oddEvenList(make_list([1,2]))) == [1,2]

from typing import Optional

class ListNode:
    def __init__(self, val: int = 0, next: 'Optional[ListNode]' = None):
        self.val = val
        self.next = next

class Solution:
    def oddEvenList(self, head: Optional[ListNode]) -> Optional[ListNode]:
        # ----------------------------------------------------------------
        # Base cases: 0 or 1 node — already trivially ordered.
        # Also guards against head.next being None below.
        # ----------------------------------------------------------------
        if head is None or head.next is None:
            return head

        # odd  = running tail of the odd-indexed sublist  (starts at node 1)
        # even = running tail of the even-indexed sublist (starts at node 2)
        # even_head is locked in now and never changes — we need it to
        # splice the two chains after the loop.
        odd       = head          # node 1  (1-based, odd)
        even      = head.next     # node 2  (1-based, even)
        even_head = even          # save forever

        # Loop invariant entering each iteration:
        #   odd  points to the last odd-indexed node processed so far
        #   even points to the last even-indexed node processed so far
        #   even.next is the next odd-indexed node (if it exists)
        #
        # We stop when even is None (list had an even number of nodes and
        # even fell off the end) OR even.next is None (even was the last
        # node, so there is no following odd node to pull over).
        while even is not None and even.next is not None:
            # Step 1: pull the next odd node (even.next) into the odd chain.
            odd.next = even.next   # odd tail now points to the upcoming odd node
            odd      = odd.next    # advance odd tail to that node

            # Step 2: pull the next even node (odd.next after advancing) into
            # the even chain.  odd.next is now the node that comes right after
            # the node we just made the odd tail — which is an even-indexed node.
            even.next = odd.next   # even tail now points to the upcoming even node
            even      = even.next  # advance even tail to that node

        # Stitch: last odd node points to the head of the even sublist.
        odd.next = even_head

        return head

# ---- helpers ----
def make_list(vals):
    dummy = ListNode(0)
    cur = dummy
    for v in vals:
        cur.next = ListNode(v)
        cur = cur.next
    return dummy.next

def list_vals(head):
    out = []
    while head:
        out.append(head.val)
        head = head.next
    return out

# ---- exhaustive tests ----
sol = Solution()

# LC examples
assert list_vals(sol.oddEvenList(make_list([1,2,3,4,5])))       == [1,3,5,2,4]
assert list_vals(sol.oddEvenList(make_list([2,1,3,5,6,4,7])))   == [2,3,6,7,1,5,4]

# Edge: empty
assert list_vals(sol.oddEvenList(make_list([])))                 == []

# Edge: single node
assert list_vals(sol.oddEvenList(make_list([42])))               == [42]

# Edge: exactly two nodes (even-length, minimal)
assert list_vals(sol.oddEvenList(make_list([1,2])))              == [1,2]

# Edge: three nodes (odd-length, the loop runs exactly once)
assert list_vals(sol.oddEvenList(make_list([1,2,3])))            == [1,3,2]

# Edge: four nodes (even-length)
assert list_vals(sol.oddEvenList(make_list([1,2,3,4])))          == [1,3,2,4]

# Larger
assert list_vals(sol.oddEvenList(make_list([1,2,3,4,5,6])))      == [1,3,5,2,4,6]
assert list_vals(sol.oddEvenList(make_list([1,2,3,4,5,6,7])))    == [1,3,5,7,2,4,6]

print("All tests passed.")

from typing import Optional

class ListNode:
    def __init__(self, val: int = 0, next: 'Optional[ListNode]' = None):
        self.val = val
        self.next = next

class SolutionDummy:
    def oddEvenList(self, head: Optional[ListNode]) -> Optional[ListNode]:
        odd_dummy  = ListNode(0)   # sentinel for odd chain
        even_dummy = ListNode(0)   # sentinel for even chain
        odd_tail   = odd_dummy
        even_tail  = even_dummy

        idx  = 1
        node = head
        while node:
            nxt = node.next    # save before we clobber node.next
            node.next = None   # cleanly detach
            if idx % 2 == 1:
                odd_tail.next = node
                odd_tail      = odd_tail.next
            else:
                even_tail.next = node
                even_tail      = even_tail.next
            node = nxt
            idx += 1

        # Splice even chain onto odd chain
        odd_tail.next = even_dummy.next
        return odd_dummy.next

# ---- helpers ----
def make_list(vals):
    dummy = ListNode(0)
    cur = dummy
    for v in vals:
        cur.next = ListNode(v)
        cur = cur.next
    return dummy.next

def list_vals(head):
    out = []
    while head:
        out.append(head.val)
        head = head.next
    return out

sol = SolutionDummy()
assert list_vals(sol.oddEvenList(make_list([1,2,3,4,5])))     == [1,3,5,2,4]
assert list_vals(sol.oddEvenList(make_list([2,1,3,5,6,4,7]))) == [2,3,6,7,1,5,4]
assert list_vals(sol.oddEvenList(make_list([])))               == []
assert list_vals(sol.oddEvenList(make_list([1])))              == [1]
assert list_vals(sol.oddEvenList(make_list([1,2])))            == [1,2]
print("SolutionDummy: all tests passed.")

from typing import Optional

class ListNode:
    def __init__(self, val: int = 0, next: 'Optional[ListNode]' = None):
        self.val = val
        self.next = next

class SolutionRecursive:
    """
    NOT O(1) space. O(n) call stack. Shown for conceptual completeness only.
    """
    def oddEvenList(self, head: Optional[ListNode]) -> Optional[ListNode]:
        # Collect all odd-indexed nodes, then all even-indexed nodes via recursion.
        odds  = self._pick(head,      skip=2)   # start at node 1, step 2
        evens = self._pick(head.next if head else None, skip=2)

        # Stitch
        tail = odds
        while tail and tail.next:
            tail = tail.next
        if tail:
            tail.next = evens
        return odds

    def _pick(self, node: Optional[ListNode], skip: int) -> Optional[ListNode]:
        """Return a new chain containing every skip-th node starting from node."""
        if node is None:
            return None
        # The node after our current one in the same parity group:
        node.next = self._pick(node.next.next if node.next else None, skip)
        return node

Source: darkinterview.com (real Reddit interview report, captured May 2026)

Problem Overview

This round is a fundamentals-heavy Machine Learning Engineer discussion. The interviewer typically starts with a simple supervised learning setup, then uses a plot of two overlapping class distributions to probe how well you understand modeling assumptions, decision boundaries, feature choice, and error trade-offs.

The question often feels conversational rather than rigidly scripted. Expect the interviewer to give hints or feedback and use your answer to branch into deeper follow-ups.

Core Prompt

You are given:

• a feature x
• a target label y

The interviewer asks: how would you model the relationship between x and y?

After that, they show a figure with two class-conditional distributions over the same feature space. The two curves overlap in the middle region. Use that figure to explain:

• whether the classes are linearly separable
• whether a single threshold is reasonable
• what the overlap means for Bayes error and unavoidable misclassification
• how you would choose a threshold in practice
• which metrics you would optimize if false positives and false negatives have different costs

What The Interviewer Is Looking For

Modeling Basics

• Clarify the prediction task first:
  • regression if y is continuous
  • binary or multiclass classification if y is categorical
• Start with a simple baseline before jumping to a more complex model
• Explain how the amount and shape of the data affect the model choice

Interpreting The Overlap Figure

The key point is not just "pick a threshold." The more complete answer is:

• if two class distributions overlap, the problem is not perfectly separable on that feature alone
• a threshold may still be a reasonable decision rule if the feature is one-dimensional and the conditional distributions are ordered
• the overlap region implies irreducible error if only this feature is used
• moving the threshold trades off precision and recall, or false positive and false negative rates
• the best threshold depends on the product goal, class prior, and error cost

Strong candidates usually go one step further and say that if performance is inadequate, they would look for additional features that better separate the classes rather than over-focusing on the threshold itself.

Common Follow-Ups

Feature Selection

The interviewer may repeatedly ask how you would choose or create better features. Be ready to discuss:

• which raw features are likely predictive
• how to identify leakage
• how to handle missing values
• scaling or normalization when needed
• encoding for categorical features
• whether interactions or nonlinear transforms could help
• how to evaluate feature importance or feature usefulness

Model Choice

You may be asked:

• when logistic regression is sufficient
• when tree-based models are a better fit
• how to think about linear vs nonlinear boundaries
• how to balance interpretability against predictive performance

Evaluation

Expect discussion around:

• train/validation/test split
• ROC-AUC vs PR-AUC for imbalanced data
• precision/recall/F1 trade-offs
• calibration and probability interpretation

Cold Start

The round may end with cold start questions, such as:

• how to make predictions for a new user with little or no history
• how to handle a new item with limited interaction data
• what fallback features or priors you would use
• when heuristics, popularity baselines, or contextual features are appropriate before enough personalized data arrives

A Strong Interview-Safe Answer Structure

1. Clarify what y represents and whether this is regression or classification.
2. Start with a simple baseline model and explain why.
3. Interpret the overlapping distributions as evidence that one feature alone cannot perfectly separate the classes.
4. Explain threshold selection in terms of business cost and metric trade-offs, not just geometry.
5. Propose additional features that could reduce overlap.
6. Discuss evaluation and how you would validate that the new features or model actually help.
7. Close with practical cold-start strategies.

Interview Experience

Candidates reported that the interviewer asked broad ML fundamentals questions throughout, gave feedback during the discussion, and seemed to care a lot about how candidates reasoned about feature selection.

One notable part of the round involved a plot where the class distributions overlapped. A shallow answer like "the data is linearly separable and we can pick a threshold" was not enough on its own. The stronger interpretation is that the overlap indicates ambiguity on the current feature, so the discussion should include irreducible error, threshold trade-offs, and what additional features might help.

Source: darkinterview.com (real Reddit interview report, captured May 2026)

Problem Overview

This Reddit Machine Learning Engineer interview is a practical tabular modeling exercise done in a Jupyter notebook. You are given a clean JSON dataset where each row represents a post impression for a user, and your goal is to predict whether the user will click the post.

The task is intentionally standard. The interviewer is mainly looking for whether you can move through a basic supervised learning workflow cleanly under time pressure:

• load JSON data
• convert it into a pandas DataFrame
• inspect the schema
• preprocess the features
• train a baseline and a few reasonable models
• evaluate the results
• explain the trade-offs behind your choices

Candidates reported that the dataset was already quite clean:

• no missing values
• no major class imbalance
• straightforward numeric and categorical features

That means the round tends to focus less on messy data cleaning and more on modeling judgment.

Core Prompt

You are given a dataset with these columns:

• hours_spent_reading_a (float)
• hours_spent_reading_b (float)
• hours_spent_reading_c (float)
• current_post_category (A, B, or C)
• click (binary label)

Interpretation:

• the first three columns describe how much time the user has historically spent reading posts in each category
• current_post_category is the category of the post currently being shown
• click indicates whether the user clicked that post

Build a model to predict click.

Assume the raw data is provided as JSON and should first be loaded and converted into a DataFrame before modeling.

Expected Workflow

Step 1: Load And Inspect The Data

The interviewer likely expects something close to:

import json
import pandas as pd

with open("data.json", "r") as f:
    data = json.load(f)

df = pd.DataFrame(data)
print(df.head())
print(df.dtypes)
print(df["click"].value_counts())

At this stage, a strong candidate should quickly confirm:

• target column is binary classification
• numeric columns are already usable
• the category feature needs encoding
• the dataset is clean enough to move straight into modeling

Step 2: Prepare Features

A reasonable preprocessing plan is:

• separate features and target
• one-hot encode current_post_category
• use train/test or train/validation/test split
• optionally standardize features for linear models

For example:

from sklearn.model_selection import train_test_split

X = df.drop(columns=["click"])
y = df["click"]

X = pd.get_dummies(X, columns=["current_post_category"], drop_first=False)

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

Reasonable Model Choices

Candidates reported comparing several common models:

• DummyClassifier for a sanity-check baseline
• logistic regression as a fast, interpretable baseline
• random forest for nonlinear interactions
• XGBoost for stronger boosted-tree performance

That is a good interview-safe progression:

• establish a trivial baseline
• train a simple linear model
• compare against tree-based nonlinear models
• discuss why one model may outperform another on this small tabular problem

What The Interviewer Is Looking For

Modeling Judgment

Be ready to explain:

• why logistic regression is a strong first baseline
• why tree-based models might capture nonlinear relationships or feature interactions
• why you did not jump immediately to a more complex model
• when interpretability matters more than raw predictive performance
• how model complexity affects speed, tuning cost, and overfitting risk

Metrics

Candidates reported direct follow-ups on metrics, including why to use one metric and not another.

Since the data appears clean and roughly balanced, acceptable primary metrics could include:

• accuracy
• F1
• ROC-AUC

A strong answer explains that the final choice should depend on product goals:

• use accuracy if classes and error costs are balanced
• use precision or recall if one mistake type matters more
• use F1 when you want a balance between precision and recall
• use ROC-AUC when comparing ranking quality across thresholds

Strong candidates also note that accuracy alone can be insufficient if business costs are asymmetric.

What To Do With More Time

The interviewer may ask what else you would do if the interview were not time-boxed. Good answers include:

• cross-validation for more robust comparison
• hyperparameter tuning
• threshold tuning
• feature engineering, such as interaction terms between historical reading profile and current category
• calibration analysis if predicted probabilities matter
• error analysis by category

A Plausible Interview-Safe Solution Structure

1. Load the JSON and convert it to a DataFrame.
2. Confirm this is a binary classification problem.
3. One-hot encode the current post category.
4. Split the data and train a DummyClassifier.
5. Train logistic regression as the main baseline.
6. Compare with random forest and XGBoost.
7. Report metrics and explain the trade-offs.
8. If time remains, mention cross-validation and hyperparameter tuning as next steps.

Common Follow-Ups

Expect questions like:

• Why did you choose logistic regression first?
• Why might random forest or XGBoost perform better here?
• Why not use SVM, k-NN, or a neural network?
• What trade-offs do you get with each model?
• Which metric would you optimize and why?
• If you had another hour, what would you do next?

Interview Experience

Candidates reported that the interview experience was friendly and practical. The interviewer allowed syntax lookups, gave small hints when the candidate made a typo, and focused on standard scikit-learn knowledge rather than deployment or monitoring.

The round was described as time-constrained but fair. Even though candidates sometimes ran out of time for cross-validation or hyperparameter tuning, the interviewer did not seem to penalize that as long as the modeling choices and reasoning were sound.

Source: darkinterview.com (real Reddit interview report, captured May 2026)

These two are the ML-system-design questions in Reddit's bank. Note what candidates report interviewers actually dwell on: for Video Recommendation it's the infra/logging/observability (event collection, pipelines) more than the model; for Feature Store it's the production side (training-serving consistency, CI/CD, rollback, caching, reliability), which candidates under-cover. Pair with the full ML system design page (Design Reddit Feed + Kafka/cache/GraphQL primers).

1. Requirements & Metrics

Functional requirements (scope clarification you should do out loud): Are we ranking only videos, or also GIFs and embeds? Answer: treat anything with a duration signal as a video. Do we serve the home feed, the dedicated video feed, or both? Answer: design for both, they share the same stack. Is integrity (NSFW, misinformation, vote manipulation) in scope? Answer: yes — Reddit's community rules make this non-negotiable.

Business metrics (North Star):

• Weekly Active Video Viewers (WAVV) — the primary growth lever.
• Session watch-time per DAU — proxy for depth of engagement.
• D7 video retention — users who watch a video today return to watch more within 7 days.

ML proxy metrics (what the model is actually trained to predict):

• pCTR — probability of click/play given impression.
• pWatch — expected watch fraction (0–1), capped at 1.0 to handle loops.
• pComplete — probability of watching >80% of duration (strong satisfaction signal).
• pDislike — probability of explicit downvote or "hide" action (integrity guard).

Engagement-vs-integrity tradeoff: A short, provocative clip maximizes pCTR but may increase pDislike and harm community trust. The ranking score must penalize integrity violations explicitly, not just hope the model learns the tradeoff implicitly.

Latency budget (state this early): Home feed P99 < 200ms end-to-end. Video player queue (pre-fetch) < 500ms. This forces a two-stage funnel — exhaustive ranking of all videos in the corpus (hundreds of millions) is impossible within budget.

2. The Two-Stage Funnel: Candidate Generation → Ranking → Re-Rank

Why a funnel? With O(100M) videos in the corpus and O(100ms) to respond, you cannot score every video with the full ranking model. The funnel uses cheap-but-high-recall retrieval to get from 100M → ~1,000 candidates, then expensive-but-precise ranking to get from 1,000 → ~50, then deterministic re-ranking rules to produce the final slate of ~20.

Stage 1: Candidate Generation (100M → ~1,000)

Two-tower retrieval model: The query tower encodes the user; the item tower encodes the video. Both towers produce 128-dimensional L2-normalized embeddings. At query time, approximate nearest-neighbor (ANN) search (Faiss HNSW index) retrieves the top-K videos by inner product. Training uses in-batch negatives plus hard negatives mined from "impressions not clicked."

The user tower input features: user ID embedding (learned), community membership bag-of-words (subreddits weighted by posting/commenting frequency), watch history sequence (last 50 videos, mean-pooled), device type, hour-of-day, day-of-week. The video tower input features: video ID embedding, subreddit ID, audio/visual embedding from a frozen video encoder (e.g., a distilled CLIP applied to keyframes), title token embeddings (mean-pooled), duration bucket, age-of-post in hours.

Multiple candidate sources — don't rely on a single retrieval path:

• ANN two-tower (~500 candidates): personalized, covers long-tail content a user has signaled interest in.
• Community trending (~200 candidates): top-K videos by upvote velocity in subscribed subreddits in the last 6h. No model needed — a simple sorted set in Redis suffices.
• Social graph (~100 candidates): videos upvoted/commented-on by Redditors the user frequently co-engages with (implicit social signal, since Reddit has no "follow" for users in general).
• Explore bucket (~200 candidates): random sample from communities the user does NOT subscribe to but is statistically similar to users who do, seeded by community embedding similarity. This combats filter-bubble formation.

Candidates from all sources are deduplicated by video ID before passing to Stage 2.

Stage 2: Ranking (~1,000 → ~50)

The ranker is a multi-task deep neural network. It takes as input: the user embedding (from the two-tower user tower, or recomputed from the feature store), the video embedding, and a rich set of cross-features that the two-tower deliberately omits (cross-features are expensive to compute for 100M items but cheap for 1,000).

Cross-features added at ranking time: user-video community overlap (does this video's subreddit match user's top communities?), recency of post relative to user's last session, video duration vs. user's typical watch duration distribution (a 30-minute video for a user whose median watch is 90 seconds is a bad recommendation), title sentiment alignment with user's engagement history.

Multi-task loss: The ranker predicts multiple targets jointly. A shared bottom MLP produces a shared representation; task-specific heads branch off:

L_total = w_ctr * BCE(p̂_ctr, y_ctr)
        + w_watch * MSE(p̂_watch, y_watch)
        + w_complete * BCE(p̂_complete, y_complete)
        + w_dislike * BCE(p̂_dislike, y_dislike)

Weights are tuned offline via a grid search on a held-out validation set optimizing for D7 retention (the North Star), not for any single task. Typical values: w_ctr ≈ 0.2, w_watch ≈ 0.4, w_complete ≈ 0.3, w_dislike ≈ 0.1 — emphasizing watch quality over raw click rate.

Final ranking score:

score = p̂_watch * p̂_complete * duration_minutes - λ * p̂_dislike

The λ * p̂_dislike penalty is a hard integrity safeguard — if the dislike probability is high (e.g., a borderline-policy video), the score is depressed regardless of watch-time. λ is a policy hyperparameter set by Trust & Safety, not tuned by the ML team.

Stage 3: Re-Ranking (~50 → ~20, deterministic rules)

Re-ranking applies deterministic business logic on top of ranked scores. It is intentionally NOT a learned model — it encodes policy, which should be auditable and adjustable without retraining.

• Diversity: No more than 2 consecutive videos from the same subreddit. Implemented as a greedy "spread" pass over the ranked list.
• Deduplication: Remove near-duplicate videos (perceptual hash similarity > 0.95 against the user's watch history in the past 30 days).
• Integrity hard filter: Remove any video with an active policy violation flag (from a separate content classifier). This is a hard removal, not a score penalty — even a score of zero is not safe enough for a policy-violating video.
• Freshness injection: Force at least 2 of the top-10 slots to be posts <4h old to prevent the feed from stagnating on popular-but-old content.
• NSFW gating: If user has not opted into NSFW content in their account settings, filter at this stage (not at retrieval, to keep the retrieval index simpler).

3. The Feedback Loop in Detail (Deepest Section)

This is what the Reddit interviewer actually wants. The model is only as good as the quality of its training labels, and those labels come entirely from the event log. Every bias in the log becomes a bias in the model.

Event Collection Architecture

Every client (web, iOS, Android) emits a stream of events to a Kafka topic. The canonical events are:

• impression: {user_id, video_id, position_in_feed, timestamp, request_id, model_version, context: "home_feed" | "video_feed" | "search"}
• play_start: {user_id, video_id, timestamp, autoplay: bool}
• heartbeat: emitted every 5 seconds of continuous watch — {user_id, video_id, elapsed_seconds, timestamp}. Used to reconstruct watch fraction without relying on a single "play_end" event (which is unreliable on mobile when app is backgrounded).
• play_end: {user_id, video_id, watched_seconds, reason: "completed" | "scroll_away" | "close_app" | "next_video"}
• upvote / downvote: {user_id, video_id, timestamp}
• share / save: {user_id, video_id, timestamp}
• hide / report: {user_id, video_id, reason, timestamp}

Critical detail: log the request_id and model_version on every impression. This is how you join training labels back to the model that generated them. Without this, you cannot do counterfactual evaluation, you cannot detect when a model update shifts the distribution, and you cannot do holdback experiments cleanly.

Watch-fraction computation from heartbeats: Because mobile apps are unreliable at sending play_end, reconstruct watch fraction from heartbeats: watch_fraction = count(heartbeats) * 5s / video_duration_s. Cap at 1.0 for looping videos. Tag reconstruction as "estimated" when play_end is missing, so downstream training can apply a lower confidence weight.

Position Bias and IPS Correction

This is the most important bias to understand. A video shown at position 1 in the feed gets clicked far more often than the same video at position 10 — not because it's better, but because users scroll less. If you train naively on raw click labels, the model learns to predict "was this shown at the top?" more than "is this a good video?" The result: items that were historically ranked high get higher pCTR predictions, which gets them ranked high again. This is a popularity feedback loop that amplifies early accidents into permanent dominance.

Correction via Inverse Propensity Scoring (IPS): For each training example (user, video, position), weight the loss by the inverse of the propensity — the probability that this video would have been shown at this position under a random policy:

Weighted loss = (1 / propensity(position)) * BCE(p̂_ctr, y_ctr)

Estimate propensity(position) empirically from a randomization bucket: for ~1% of traffic, randomize the ranking order before serving. The empirical click rate at each position in that bucket gives you the position-click curve. Fit a simple exponential decay: propensity(k) = α * exp(-β * k). Typical values: position 1 → propensity 0.30, position 5 → 0.18, position 10 → 0.09.

IPS correction has variance problems at extreme positions (very high weights for low positions). Clip weights at a maximum of 10x to prevent a few low-position impressions from dominating the loss.

Label Leakage

Label leakage occurs when features available at training time contain information that would not be available at serving time. In a recommendation system, the subtlest form is future engagement leakage: if your training pipeline computes "video total_upvotes" as of today, and the model trains on an impression that happened 3 months ago, the feature includes 3 months of future upvotes that the model could not have seen at serving time. The model learns to recommend videos that became popular, not videos that were about to become popular.

Prevention: point-in-time correct (PIT-correct) features. Every feature used in training must be computed as of the timestamp of the impression event, not as of when the training pipeline runs. Implementation:

1. Store the full time-series of every feature in the feature store (or warehouse), not just the current value.
2. The training pipeline joins impression events to features using an AS-OF join: SELECT * FROM impressions i JOIN video_features f ON i.video_id = f.video_id AND f.feature_ts <= i.event_ts ORDER BY f.feature_ts DESC LIMIT 1.
3. For user features (watch history), reconstruct the user's history as it existed at the time of the impression — exclude any videos watched after the impression timestamp.

PIT-correct joins are expensive. Practical tradeoff: apply strict PIT-correctness for slow-moving features (community membership, user age) and accept a small amount of leakage for fast-moving features (video upvote count in the last hour) by using the value from the nearest available snapshot.

Training-Serving Skew

Training-serving skew is the condition where the feature values seen by the model during training differ from the feature values seen during inference, not due to distribution shift (a legitimate change over time) but due to engineering inconsistency (computing features differently in two code paths). It is one of the most common and hardest-to-detect bugs in production ML systems.

Sources of skew in a recommendation system:

• The training pipeline computes watch_fraction from heartbeats; the serving feature store computes it from play_end events only. These two paths give different numbers for the same video.
• Text features (video title embeddings) are tokenized with a different vocabulary version in the training pipeline vs. the online model server.
• Timestamp-based features (video age in hours) are computed relative to the training pipeline's run time, not the impression timestamp.
• Feature normalization (z-score of video duration) uses statistics computed on the training set; at serving time, you apply those same statistics to new data — fine — but if you forget to apply them at all, the model sees out-of-distribution inputs.

Detection: Log the feature vector used at serving time alongside the impression event (or a hash of it). In the training pipeline, after computing features for a training example, compare to the logged serving-time vector. Compute the mean absolute deviation between the two for each feature. Alert if any feature's MAD exceeds a threshold (e.g., >5% of the feature's standard deviation). Run this check on a 0.1% sample of impressions — logging the full feature vector for 100% of impressions is too expensive.

Closing the Training Loop

Training data pipeline (daily batch, with near-real-time shadow):

1. Kafka impression and engagement events land in the data warehouse (e.g., Iceberg tables on S3) via a Flink consumer. Raw events are available with ~5-minute latency.
2. A daily Spark job applies the label join: for each impression, join to engagement events within a 48h window (allow 48h for "lazy" signals — a user might come back and upvote the next day). Apply IPS weights. Reconstruct PIT-correct features. Output a labeled training dataset in columnar format (Parquet).
3. A model training job (PyTorch + distributed training on 8–16 GPUs) trains on the last 30 days of labeled data, with down-sampling of older data (exponential time-decay weighting — a 30-day-old example gets weight 0.3x a today example). Training takes ~4–6 hours.
4. The new model is evaluated offline on a held-out validation set (last 2 days, not included in training) against AUC-ROC for pCTR, RMSE for pWatch. It must beat the current production model on both by >0.2% (a statistically significant threshold determined by bootstrap resampling).
5. Passing validation triggers a canary deployment (1% of traffic). Shadow evaluation runs for 24h. If no degradation in online metrics, ramp to 10% → 50% → 100% over 3 days.

4. Serving Architecture & Feature Store

Request path (target P99 < 200ms for home feed):

1. Client request hits the Feed Gateway (load balancer). ~2ms.
2. Gateway calls the Recommendation Service, passing user_id, context, device, session_id.
3. Recommendation Service fetches user features from the online feature store (Redis/DynamoDB) in parallel. ~5ms.
4. Recommendation Service fans out to three candidate sources in parallel: ANN retrieval service (HNSW lookup), trending API (Redis sorted set), social graph service. Waits for all three with a 30ms timeout; any source that doesn't respond is dropped. ~20–30ms total.
5. Recommendation Service calls the Ranking Service with merged candidates + user features. ~50ms for DNN scoring of ~1,000 items (vectorized batch inference).
6. Re-ranking rules applied in-memory. ~5ms.
7. Response assembled and returned. ~2ms serialization.

Total: ~85ms median, ~150ms P95, ~200ms P99. Leaves headroom for network jitter.

Online feature store design: Two tiers. Tier 1: ultra-low-latency (Redis, <1ms) for the ~50 features used at every request: user watch history (last 50 video IDs, stored as a list), subscribed subreddits (list of subreddit IDs), user's embedding vector (128 floats), and per-video statistics (upvote count, view count, age). Tier 2: moderate-latency (DynamoDB, ~5ms) for less frequently changing features: user community embedding, user device history, long-term watch preferences.

Candidate caching: For non-personalized retrieval (trending), cache the top-200 trending videos per subreddit in Redis with a 60-second TTL. This avoids re-running the sorted set query on every request for popular subreddits. For personalized ANN results, caching is not worth it — user embeddings change fast enough that a 60-second stale cache degrades quality measurably.

Pre-fetching for the video player queue: When a user starts watching a video, the client immediately requests the next 5 videos in the queue in a background call. This relaxes the latency budget to 500ms for the queue request, allowing a deeper ranking pass (more candidates, more features). The pre-fetch result is stored client-side and served instantly when the user swipes to the next video.

5. Cold Start

New User Cold Start

A new Reddit account has no watch history, no upvotes, no community memberships. The ANN two-tower returns nothing meaningful for this user.

Strategy — onboarding signals: At account creation, prompt the user to select 5+ interest topics (not just subreddits — higher-level tags like "gaming," "cooking," "news"). Map these tags to communities. Use the community embedding to construct an initial user embedding as a weighted average of community embeddings. This gives the two-tower something to work with on the first session.

Strategy — session bootstrapping: Treat the first session as an exploration phase. Override the ranking score with a multi-armed bandit policy (Thompson Sampling) that explores content from the selected topics with high variance. Update the bandit's posterior after each engagement event within the session (online update, no model retraining). After 5–10 interactions in the session, the user's implicit interest profile is strong enough to switch to the normal ranking model.

Strategy — demographic proxy (with caution): If the user provides location or device locale at signup, use the aggregate engagement patterns of users from that region/locale as a prior. This is useful but must be audited for fairness — do not proxy demographic attributes that correlate with protected characteristics.

New Video Cold Start

A newly posted video has zero engagement history. The two-tower item embedding is initialized from content features only (title, thumbnail, subreddit, duration). The ranking model's pCTR and pWatch predictions are based entirely on content features, which are noisier than engagement-calibrated features.

Strategy — exploration injection: Every video, for its first 2 hours, is eligible for inclusion in the "explore bucket" of candidate generation. It gets shown to ~500 sampled users whose community memberships match the video's subreddit. These exposures generate initial engagement signal.

Strategy — warm-up prior: Initialize pCTR and pWatch for a new video to the median values for videos of the same subreddit and duration bucket from the last 30 days. This prevents extreme over-ranking (if the model is uncertain, it defaults to median performance, not zero). Concretely: replace missing engagement features with subreddit-duration-bucket means, flagged as "cold_start=true" in the feature vector so the model can learn to apply a different weight.

Strategy — content-based fallback: Run a separate content-only ranker (trained solely on content features, no engagement features) for cold-start items. Blend the content-only score with the engagement-based score using a mixing weight that transitions from 1.0 (content-only) to 0.0 (engagement-based) as the video accumulates impressions: score = α(n) * content_score + (1 - α(n)) * engagement_score, where α(n) = exp(-n/50) and n is the impression count.

6. Online Evaluation: A/B Testing, Holdbacks, Guardrails

A/B testing framework: User-level assignment (not session-level) — a user is assigned to control or treatment for the duration of the experiment to avoid carry-over effects. Assignment is deterministic (hash of user_id + experiment_id) so the same user always sees the same variant. Experiments run for a minimum of 2 weeks to capture weekly periodicity in engagement (Reddit has strong Monday-vs-weekend usage differences).

Primary and guardrail metrics: Every experiment defines one primary metric (e.g., session watch-time) and several guardrail metrics (e.g., D7 retention, complaint rate, diversity score). An experiment "wins" only if: (1) the primary metric improves by a statistically significant margin (p < 0.05 after multiple-testing correction), AND (2) no guardrail metric degrades by more than a pre-set threshold (e.g., D7 retention does not drop by more than 0.5%).

Holdback experiments: To measure the true long-term value of the recommendation system, maintain a 1% holdback group that never sees recommended videos — they see only chronological community feeds. Compare WAVV and D7 retention between the holdback and the recommendation-treated population. This is your ground truth for "does the rec system actually help?" Run holdbacks for 90-day windows to capture long-term effects.

Online metrics instrumentation: Do not rely solely on offline metrics (AUC, RMSE) to decide if a model is good. Compute online proxy metrics directly from the event stream during the canary period: rolling 1h and 24h watch-time per impression, upvote rate, downvote rate, scroll-away rate (impression with no play event within 3 seconds), and session end rate (user leaves the app within 60 seconds of seeing a video). Alert if any of these degrade by >2% relative in the canary period.

7. Infrastructure: Kafka → Flink → Warehouse + Online Store

Kafka event bus: All client events (impressions, heartbeats, plays, engagements) publish to partitioned Kafka topics. Partition key is user_id — this ensures all events for a user are on the same partition, making stateful stream processing (e.g., computing session-level features) correct. Topic retention is 7 days (enough to replay if a downstream consumer falls behind). Peak throughput on a Reddit-scale platform: ~500K events/second, requiring ~50 Kafka partitions at 10K events/partition/second.

Flink stream processing layer:

• Session stitching job: Groups events by (user_id, session_id) and computes session-level features — total watch time in session, number of unique subreddits, number of videos played. Outputs to a session features table (Iceberg).
• Real-time feature update job: Consumes engagement events and updates the online feature store. When a play_end event arrives, compute the new watch_fraction and write it to Redis for that video_id. When an upvote arrives, increment the video's upvote counter in Redis. Flink's exactly-once processing guarantee is critical here — you cannot double-count upvotes due to consumer retries.
• Trending computation job: Computes upvote velocity per video per subreddit over a sliding 6-hour window. Maintains a top-200 sorted set in Redis for each subreddit. Uses a Flink keyedProcessFunction with a MapState keyed by (subreddit_id, video_id).
• Training data generation job (near-real-time shadow): Joins impression events to engagement events with a 48h event-time window (Flink handles out-of-order events with watermarks at 2h late arrival tolerance). Outputs labeled training examples to an Iceberg table on S3, which the batch training job reads. This is the "shadow pipeline" — it runs continuously but training consumes it daily.

Data warehouse (Iceberg on S3 + query engine like Trino): Stores the full historical event log, labeled training datasets, feature snapshots (for PIT-correct joins), model evaluation results, and experiment assignment tables. Iceberg's time-travel capability is essential for PIT-correct training: you can query SELECT * FROM video_features FOR SYSTEM_TIME AS OF '2026-03-15 14:23:00' to get the exact feature values at any past timestamp.

Feature store dual-write pattern: Flink jobs write to both the online store (Redis, for serving) and the offline store (Iceberg, for training). This dual-write architecture is the primary mechanism for preventing training-serving skew: both the serving path and the training path read features from the same source-of-truth Flink job. Any divergence in feature computation exists in one place, not in two separate codebaths that drift apart over time.

Source: darkinterview.com (real Reddit interview report, captured May 2026)

Problem Statement

Design a feature store for Reddit's ML platform. Explain how offline and online feature storage stay consistent, how features are computed and materialized, how training and inference use the same definitions, and how the platform supports safe rollout across many teams.

Comprehensive Solution Resources

For detailed feature store design walkthroughs and implementation strategies, these resources are a good starting point:

• HOW TO DESIGN A FEATURE STORE FOR YOUR MLOPS PIPELINE | ML SYSTEM DESIGN
• ML System Design: Feature Store

The notes below focus on what interviewers often emphasize in real Reddit loops, so you can spend prep time on the parts candidates most often under-cover.

What Interviewers Often Care About

This question is easy to answer too heavily from the modeling side and too lightly from the production side. Candidates often spend most of their time on feature engineering, model architecture, and high-level data flow. Stronger answers also cover:

• how feature definitions are versioned and tested before rollout
• what CI/CD looks like for feature pipelines, transformations, and schema changes
• how deployments are staged and rolled back safely
• how online caching affects latency, freshness, and consistency
• how to preserve training-serving consistency across offline and online systems
• how to handle backfills, stale data, failed jobs, and degraded upstream dependencies
• what reliability and observability metrics should guard the platform

Real Interview Experience

Candidates reported that the interviewer explicitly wanted a feature store design, but pushed beyond the usual architecture diagram. The discussion often focused on testing, deployment, CI/CD, caching, and reliability optimization, because those parts are easy to overlook when candidates focus only on modeling or feature engineering.

Source: darkinterview.com (real Reddit interview report, captured May 2026)

A second crowdsourced bank (interviewdb.io) lists 21 Reddit questions; most overlap the set above, but these six are genuinely additional and recent (reported 1 week – 6 months ago). Added with worked solutions. Candidate-reported; verify framing.

from collections import defaultdict, deque

def find_build_order(tasks, dependencies):
    graph = defaultdict(list)
    indeg = {t: 0 for t in tasks}
    for a, b in dependencies:        # a depends on b  ->  b before a
        graph[b].append(a)
        indeg[a] += 1
    q = deque([t for t in tasks if indeg[t] == 0])
    order = []
    while q:
        node = q.popleft()
        order.append(node)
        for nxt in graph[node]:
            indeg[nxt] -= 1
            if indeg[nxt] == 0:
                q.append(nxt)
    return order if len(order) == len(tasks) else []   # [] signals a cycle

def best_kiosk(points):
    xs = sorted(p[0] for p in points)
    ys = sorted(p[1] for p in points)
    return (xs[len(xs)//2], ys[len(ys)//2])   # median per axis

from collections import deque

def related_up_to_degree(community_map, follower_map, target, degree):
    seen = {target}
    frontier = {target}
    result = []
    for _ in range(degree):
        nxt = set()
        for c in frontier:
            for f in community_map[c]:        # followers of c
                for c2 in follower_map[f]:    # other communities they follow
                    if c2 not in seen:
                        seen.add(c2); nxt.add(c2); result.append(c2)
        frontier = nxt
    return result
# related_up_to_degree(cm, fm, "C4", 1) -> ["C1"]
# related_up_to_degree(cm, fm, "C4", 2) -> ["C1", "C2"]

from collections import defaultdict

def render(replies):                      # replies: [{id, parentId, text}]
    children = defaultdict(list)
    for r in replies:
        children[r["parentId"]].append(r)
    out = []
    def dfs(parent_id, depth):
        for r in children[parent_id]:     # optionally sort by score/time here
            out.append("    " * depth + r["text"])
            dfs(r["id"], depth + 1)
    dfs(None, 0)                          # roots have parentId == None
    return out

These four are in Reddit's real bank but are tagged System Design / Software Engineer, not ML. Kept brief since you're on the ML track — skim for the patterns (real-time fanout, sorted-set leaderboards, notification pipelines, tree ranking). If one comes up, the framework and infra primers on the ML system design page transfer directly. All four follow the same darkinterview 5-phase template: Requirements → Data model → API → High-level design → Scaling/trade-offs.