Full-stack engineer coding interview questions

As asked

Implement a debounce function that delays invoking a callback until after a given wait period has elapsed since the last call. It should return a cancel method to clear any pending invocation.

Sample answer outline

Use a closure to hold the timeout handle. On each call, clear any existing timeout and set a new one. The cancel method calls clearTimeout. The candidate should handle the correct this context, optional leading-edge invocation, and explain why debounce is appropriate for search input vs throttle for scroll handlers.

Reference implementation (typescript)

function debounce<T extends (...args: unknown[]) => void>(
  fn: T,
  wait: number
): T & { cancel: () => void } {
  let timer: ReturnType<typeof setTimeout> | undefined;
  const debounced = (...args: Parameters<T>) => {
    // TODO: implement
  };
  debounced.cancel = () => { /* TODO */ };
  return debounced as T & { cancel: () => void };
}

Expect these follow-ups

• How would you make this debounce also support a leading option that fires immediately on the first call?

As asked

Implement a Least Recently Used cache class in TypeScript with a fixed capacity. Both get(key) and put(key, value) must run in O(1) time.

Sample answer outline

Combine a doubly-linked list with a Map. The Map gives O(1) lookup of nodes; the doubly-linked list maintains access order. On get, move the node to the front. On put, insert at the front; if over capacity, remove from the tail. JavaScript's built-in Map preserves insertion order and can be used as a lightweight alternative by deleting and re-inserting on access, though the interviewee should discuss the tradeoffs.

Reference implementation (typescript)

class LRUCache {
  private capacity: number;
  // hint: use a Map and a doubly-linked list
  constructor(capacity: number) {
    this.capacity = capacity;
  }
  get(key: number): number {
    // return -1 if not found
    return -1;
  }
  put(key: number, value: number): void {
    // evict LRU entry if at capacity
  }
}

Expect these follow-ups

• How would you make this LRU cache safe for concurrent access in a multi-threaded environment?

As asked

Given a table employees(id, name, department_id, salary), write a SQL query that returns each department's second highest salary. If a department has fewer than two distinct salaries, exclude it from the results.

Sample answer outline

Use a window function: DENSE_RANK() OVER (PARTITION BY department_id ORDER BY salary DESC) to rank salaries within each department, then filter for rank = 2 in a CTE or subquery. The candidate should know the difference between RANK, DENSE_RANK, and ROW_NUMBER (DENSE_RANK handles ties correctly here), and should be able to write the equivalent without window functions using a correlated subquery or self-join.

Reference implementation (sql)

-- employees(id, name, department_id, salary)
-- Write a query returning (department_id, second_highest_salary)
-- Exclude departments with fewer than 2 distinct salaries

SELECT department_id, salary AS second_highest_salary
FROM (
  SELECT
    department_id,
    salary,
    DENSE_RANK() OVER (
      PARTITION BY department_id
      ORDER BY salary DESC
    ) AS rnk
  FROM employees
) ranked
WHERE rnk = 2;

Expect these follow-ups

• How would you rewrite this without window functions for a database that does not support them?

As asked

Given a string, find the first character that does not repeat anywhere in the string. If no such character exists, return an empty character or null. Optimize for a single pass through the string.

Sample answer outline

Two-pass approach: first pass builds a frequency hashmap, second pass scans left to right and returns the first character with count 1. A true single-pass approach uses an ordered map or a list of (char, first_index) pairs updated as you scan. A strong candidate mentions character set assumptions (ASCII vs Unicode), handles uppercase vs lowercase, and asks whether the empty string should raise or return a sentinel.

Reference implementation (python)

def first_unique_char(s: str) -> str:
    count = {}
    for ch in s:
        count[ch] = count.get(ch, 0) + 1
    for ch in s:
        if count[ch] == 1:
            return ch
    return ''

Expect these follow-ups

• What if the string is too large to fit in memory and arrives as a stream?
• Extend this to find the first character that appears exactly K times.

As asked

Airbnb's listing availability service caches per-listing calendar data to avoid hitting the database on every search request. Implement an LRU (Least Recently Used) cache that supports get and put in O(1) time. Then explain how you would extend it to be thread-safe in a Node.js or Java service.

Sample answer outline

The canonical solution uses a doubly linked list for O(1) removal combined with a hash map for O(1) lookup. The candidate should walk through get (move node to head), put (insert at head, evict tail if at capacity), and explain pointer bookkeeping. For thread safety they should discuss synchronizing the composite operation, or using a read-write lock.

Reference implementation (python)

class LRUCache:
    def __init__(self, capacity: int):
        self.cap = capacity
        self.cache = {}  # key -> node
        # sentinel head and tail
        self.head, self.tail = Node(0, 0), Node(0, 0)
        self.head.next = self.tail
        self.tail.prev = self.head

    def get(self, key: int) -> int:
        if key not in self.cache:
            return -1
        self._move_to_head(self.cache[key])
        return self.cache[key].val

    def put(self, key: int, value: int) -> None:
        if key in self.cache:
            node = self.cache[key]
            node.val = value
            self._move_to_head(node)
        else:
            node = Node(key, value)
            self.cache[key] = node
            self._add_to_head(node)
            if len(self.cache) > self.cap:
                tail = self._remove_tail()
                del self.cache[tail.key]

Expect these follow-ups

• What changes if you need the cache to be distributed across multiple servers?
• How would you add a TTL expiry to each entry without breaking O(1) operations?

As asked

Airbnb uses tree structures internally, for example to represent hierarchical location data (continent, country, city, neighborhood). Implement serialization and deserialization of a binary tree. The serialized form can be any format you choose, as long as the deserialized tree is identical to the original. Discuss your format choice before coding.

Sample answer outline

A clean solution uses BFS with null markers, encoding as a comma-separated string. The candidate should explain why they chose BFS over DFS (often simpler to reconstruct), handle null nodes explicitly, and walk through reconstruction using a queue of values. Edge cases include single-node trees and all-null children.

Reference implementation (python)

from collections import deque

def serialize(root):
    if not root:
        return 'null'
    result, queue = [], deque([root])
    while queue:
        node = queue.popleft()
        if node:
            result.append(str(node.val))
            queue.append(node.left)
            queue.append(node.right)
        else:
            result.append('null')
    return ','.join(result)

def deserialize(data):
    vals = data.split(',')
    if vals[0] == 'null':
        return None
    root = TreeNode(int(vals[0]))
    queue, i = deque([root]), 1
    while queue:
        node = queue.popleft()
        if vals[i] != 'null':
            node.left = TreeNode(int(vals[i]))
            queue.append(node.left)
        i += 1
        if vals[i] != 'null':
            node.right = TreeNode(int(vals[i]))
            queue.append(node.right)
        i += 1
    return root

Expect these follow-ups

• How would you change your approach if the tree had up to 10 million nodes and you needed to minimize serialized size?
• Can you serialize it in a way that lets you deserialize any subtree without reading the whole string?

As asked

Design a parking lot system that handles multiple levels, different vehicle types (motorcycle, car, bus), and can tell you whether a spot is available. Walk me through your class design first, then write the core classes.

Sample answer outline

A strong answer starts with clarifying questions about scale, pricing, and vehicle types before writing code. Core classes should include ParkingLot, Level, ParkingSpot (with a type enum), and Vehicle (with a size property). The candidate should use polymorphism so different vehicle types check availability correctly, and should handle the edge case where a bus takes multiple spots. Interviewers look for clean separation of concerns and good use of OOP principles, not just a working sketch.

Reference implementation (python)

class ParkingSpot:
    def __init__(self, spot_type: str):
        self.spot_type = spot_type  # 'compact', 'large', 'motorcycle'
        self.is_occupied = False
        self.vehicle = None

    def can_fit(self, vehicle: 'Vehicle') -> bool:
        raise NotImplementedError

    def park(self, vehicle: 'Vehicle') -> bool:
        if self.can_fit(vehicle) and not self.is_occupied:
            self.vehicle = vehicle
            self.is_occupied = True
            return True
        return False

Expect these follow-ups

• How would you extend this to support reservations?
• How would you make it thread-safe if multiple entry gates are updating spot availability concurrently?

As asked

Implement a rate limiter class that allows at most N requests per time window T (in seconds) per user. It should support isAllowed(user_id, timestamp) which returns true if the request is allowed and false if it should be throttled. Timestamps are in seconds and may arrive slightly out of order.

Sample answer outline

A strong answer implements a sliding window using a deque or sorted list of recent timestamps per user. For each request, remove timestamps older than (current_time - T) from the window, then check if the window size is below N before adding the current timestamp. Time complexity is O(N) per call in the worst case. The candidate should contrast this with a fixed window counter (simpler but has edge-case burst at window boundaries) and a token bucket (better for bursty traffic, harder to implement exactly).

Reference implementation (python)

from collections import defaultdict, deque

class RateLimiter:
    def __init__(self, n: int, t: int):
        self.n = n
        self.t = t
        self.windows: dict[str, deque] = defaultdict(deque)

    def is_allowed(self, user_id: str, timestamp: int) -> bool:
        window = self.windows[user_id]
        cutoff = timestamp - self.t
        while window and window[0] <= cutoff:
            window.popleft()
        if len(window) < self.n:
            window.append(timestamp)
            return True
        return False

Expect these follow-ups

• How would you implement this rate limiter in a distributed system where requests for the same user hit different servers?
• How does a token bucket differ from a sliding window in terms of burst handling?

As asked

Implement a thread-safe sliding-window rate limiter in Go. It should allow at most N requests per user per minute, use O(1) memory per user (not storing every request timestamp), and return whether the current request is allowed along with the number of remaining requests in the window.

Sample answer outline

A strong answer implements a token bucket or sliding-window counter using atomic operations or a mutex-protected struct, explains the tradeoff between the exact sliding log (O(N) memory, precise) and the sliding window counter approximation (O(1) memory, slightly imprecise near window boundaries), and handles the concurrent access correctly. The candidate should mention Redis sorted sets as the production approach for distributed rate limiting.

Reference implementation (go)

type RateLimiter struct {
    mu       sync.Mutex
    counters map[string]*windowCounter
    limit    int
    window   time.Duration
}

type windowCounter struct {
    prevCount  int
    currCount  int
    windowStart time.Time
}

func (rl *RateLimiter) Allow(userID string) (bool, int) {
    rl.mu.Lock()
    defer rl.mu.Unlock()
    // TODO: implement sliding window logic
    return false, 0
}

Expect these follow-ups

• How would you make this work across multiple API server instances?
• What happens to accuracy at the window boundary with a counter-based approach versus a log-based approach?

As asked

Implement an LRU cache that supports get(key) and put(key, value) both in O(1) time. The cache has a fixed capacity; when it is full, evict the least recently used entry before inserting a new one. Use any language you prefer.

Sample answer outline

The canonical solution uses a doubly linked list for O(1) eviction from both ends and a hash map for O(1) key lookup, with sentinel head and tail nodes to simplify boundary conditions. Strong candidates implement the move-to-front logic cleanly, handle the capacity-one edge case, and note that Python's OrderedDict gives this for free if the interviewer allows it.

Reference implementation (python)

class LRUCache:
    def __init__(self, capacity: int):
        pass

    def get(self, key: int) -> int:
        # Return -1 if not found
        pass

    def put(self, key: int, value: int) -> None:
        pass

Expect these follow-ups

• How would you make this cache thread-safe for concurrent access?
• What changes if you need the cache to be persistent across process restarts?

As asked

Given the path to a root directory, write a function that returns groups of files that have identical content. Assume the directory tree can contain millions of files and the total data can be hundreds of gigabytes. How do you do this efficiently, and how do you avoid reading every file fully?

Sample answer outline

A strong approach first groups by file size (eliminating obvious non-duplicates), then hashes only the first 4KB of each same-size group to eliminate more, then fully hashes the remaining candidates. This avoids reading most data. The candidate should mention using SHA-256 or similar, handle hash collisions gracefully, and discuss parallelising the I/O with a thread pool. Very relevant to Dropbox's actual deduplication work.

Reference implementation (python)

import os, hashlib
from collections import defaultdict

def find_duplicates(root: str) -> list[list[str]]:
    # Return list of groups; each group is a list of paths with identical content
    by_size = defaultdict(list)
    for dirpath, _, filenames in os.walk(root):
        for f in filenames:
            path = os.path.join(dirpath, f)
            by_size[os.path.getsize(path)].append(path)
    # ... continue
    pass

Expect these follow-ups

• How would you handle symbolic links and circular directory references?
• If you had to run this incrementally on a live directory that is changing, how would your approach differ?

As asked

Design an algorithm to serialize a binary tree to a string and deserialize that string back to the original tree. Your serialization format must handle any binary tree, not just BSTs. What format do you choose and why?

Sample answer outline

A strong answer uses preorder traversal with explicit null markers (e.g., 'null') so the structure is recoverable from a single pass. The candidate should implement both functions cleanly, handle empty trees, discuss why inorder alone is insufficient for reconstruction, and note that for very deep trees an iterative solution avoids stack overflow.

Reference implementation (python)

class TreeNode:
    def __init__(self, val=0, left=None, right=None):
        self.val = val
        self.left = left
        self.right = right

def serialize(root: TreeNode) -> str:
    pass

def deserialize(data: str) -> TreeNode:
    pass

Expect these follow-ups

• How does your format change if the values are arbitrary strings that might contain your delimiter character?
• Could you reconstruct the tree from level-order (BFS) serialization? What are the tradeoffs?

As asked

Given an array of integers and a target sum, return the indices of the two numbers that add up to the target. Assume exactly one solution exists. First give me a brute force approach, then optimize it. After that: if I told you the input array is sorted, how does your solution change? And finally, what if I needed you to find all unique pairs, not just one?

Sample answer outline

A strong candidate moves from O(n^2) brute force to O(n) hash map immediately and explains the space-time tradeoff. For the sorted variant they pivot to two pointers in O(n) time O(1) space. For all unique pairs they handle duplicates by skipping repeated elements at each pointer position. Google wants to see fluency in choosing the right data structure, not just knowing the answer.

Reference implementation (python)

def two_sum(nums: list[int], target: int) -> list[int]:
    seen = {}
    for i, n in enumerate(nums):
        complement = target - n
        if complement in seen:
            return [seen[complement], i]
        seen[n] = i
    return []

Expect these follow-ups

• What if the numbers could overflow a 32-bit integer?
• How would you test this function?

As asked

Lyft's API gateway needs to enforce per-user rate limits across a distributed fleet of backend servers. Implement a rate limiter that enforces a sliding-window limit of N requests per minute per user ID. You can use Redis. Write the core logic and explain your choice of data structure.

Sample answer outline

Strong answers use a Redis sorted set with timestamps as scores, add each request with ZADD, remove entries older than 60 seconds with ZREMRANGEBYSCORE, and count with ZCARD, all in a Lua script for atomicity. They also discuss the trade-off with fixed-window (simpler but bursty) and token bucket (smoother but more state).

Reference implementation (python)

import redis

def is_allowed(r: redis.Redis, user_id: str, limit: int, window_secs: int) -> bool:
    key = f"rl:{user_id}"
    now_ms = int(time.time() * 1000)
    window_start = now_ms - window_secs * 1000
    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)
    pipe.zadd(key, {str(now_ms): now_ms})
    pipe.zcard(key)
    pipe.expire(key, window_secs + 1)
    results = pipe.execute()
    count = results[2]
    return count <= limit

Expect these follow-ups

• What happens to your rate limiter if the Redis node goes down?
• How would you rate-limit by IP instead of user ID for unauthenticated endpoints?

As asked

OpenAI's inference stack needs to cache KV (key-value) states and decoded outputs to serve repeated or similar prompts efficiently. As a warm-up, implement a general LRU cache that supports two operations: get(key) returns the value if the key exists, otherwise -1; put(key, value) inserts or updates the key, evicting the least-recently-used key when the cache is at capacity. Both operations must run in O(1) time.

Sample answer outline

The candidate should reach for a doubly-linked list paired with a hash map. They should explain why a doubly-linked list allows O(1) removal (you need the prev pointer) and why the hash map stores node references, not just keys. A strong answer handles edge cases: capacity of 1, overwriting an existing key without duplicating the node, and moving the accessed node to the head correctly.

Reference implementation (python)

class LRUCache:
    def __init__(self, capacity: int):
        self.cap = capacity
        self.cache = {}  # key -> node
        # sentinel head and tail
        self.head = Node(0, 0)
        self.tail = Node(0, 0)
        self.head.next = self.tail
        self.tail.prev = self.head

    def get(self, key: int) -> int:
        ...

    def put(self, key: int, value: int) -> None:
        ...

Expect these follow-ups

• How would you make this thread-safe for concurrent access?
• What changes if you need to support a TTL on each entry in addition to LRU eviction?

As asked

Given a stream of Server-Sent Events from the Claude API where each event looks like `data: {"type": "content_block_delta", "delta": {"type": "text_delta", "text": "hello"}}`, write a function that accumulates the full text and returns the total character count once a `message_stop` event is received. Handle malformed events gracefully.

Sample answer outline

The function should iterate over SSE lines, skip comment and empty lines, parse the JSON payload after the `data: ` prefix, check the event type, and accumulate text from content_block_delta events. On message_stop it returns the accumulated string and its length. Malformed JSON should be caught and logged without crashing the accumulator.

Reference implementation (python)

def parse_sse_stream(lines: list[str]) -> tuple[str, int]:
    # lines is a sequence of raw SSE lines
    accumulated = []
    for line in lines:
        if not line.startswith('data: '):
            continue
        payload = line[len('data: '):]
        # TODO: parse payload, accumulate text, stop on message_stop
    return ''.join(accumulated), sum(len(s) for s in accumulated)

Expect these follow-ups

• How would you modify this to handle streaming tool use blocks in addition to text blocks?
• How would you test this without hitting the real API?

As asked

Write a Python function that calls a hypothetical `claude_complete` function, parses tool calls from the response, executes them, feeds results back, and retries the loop up to three times if a tool execution raises an exception. Stop when the model returns a final text response with no more tool calls.

Sample answer outline

The loop calls claude_complete with the current messages list, checks for tool_use blocks in the response content, executes each tool, appends the tool result as a user message, and iterates. On exception from a tool, it appends an error result and increments a retry counter. If retries exceed the limit, it raises or returns a failure. When the response contains no tool blocks, the text is the final answer. A strong solution handles the message format correctly and resets the retry counter on a successful tool call.

Reference implementation (python)

def run_agent(initial_messages: list[dict], tools: dict) -> str:
    messages = list(initial_messages)
    retries = 0
    max_retries = 3
    while True:
        response = claude_complete(messages=messages, tools=list(tools.values()))
        # check response.content for tool_use blocks
        # execute tools, append results
        # if no tool calls, return text
        # if tool error, increment retries or raise
        pass

Expect these follow-ups

• How would you add a timeout to individual tool calls without blocking the event loop?
• How would you log each turn of the agent loop for debugging without storing sensitive tool results?

As asked

You are writing a script that needs to pull all clients across 500 Meraki networks using the Meraki Dashboard API. The API paginates using Link headers with rel='next' URLs and has a rate limit of 10 requests per second per organization. Walk me through how you implement this correctly without hitting rate limits or losing data.

Sample answer outline

A strong answer parses the Link header to get the next page URL rather than constructing URLs manually (the rel=next URL may include a startingAfter cursor parameter). It implements token-bucket or fixed-window rate limiting at 10 RPS. For 500 networks, parallel requests with a concurrency cap and retry logic with exponential backoff on 429 responses are necessary. The candidate should mention that Meraki's X-Request-Id header can help debug failures and that some endpoints have a perPage maximum of 1000.

Reference implementation (python)

import httpx, asyncio, re
from collections import deque

async def get_all_pages(url: str, headers: dict) -> list:
    results = []
    async with httpx.AsyncClient() as client:
        while url:
            r = await client.get(url, headers=headers)
            r.raise_for_status()
            results.extend(r.json())
            link = r.headers.get('Link', '')
            match = re.search(r'<([^>]+)>;\s*rel="next"', link)
            url = match.group(1) if match else None
    return results

Expect these follow-ups

• How would you checkpoint progress so a restart does not re-fetch all pages from the beginning?
• What happens if the Link header is missing mid-stream?

As asked

Write a function that takes a CIDR notation string like '192.168.10.0/22' and returns the network address, broadcast address, first usable host, last usable host, and total number of usable hosts. Do not use ipaddress or any other networking library.

Sample answer outline

A strong answer converts the IP to a 32-bit integer, creates the subnet mask by shifting 1s left by (32 - prefix_length), ANDs the IP with the mask to get the network address, ORs the network with the inverted mask to get the broadcast, and adds/subtracts 1 for the first/last usable hosts. Total usable hosts = 2^(32 - prefix) - 2 (excluding network and broadcast). The candidate should handle edge cases: /31 (used for point-to-point links, both addresses are usable per RFC 3021) and /32 (single host).

Reference implementation (python)

def subnet_info(cidr: str) -> dict:
    ip_str, prefix = cidr.split('/')
    prefix = int(prefix)
    parts = list(map(int, ip_str.split('.')))
    ip_int = (parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3]
    mask = (0xFFFFFFFF << (32 - prefix)) & 0xFFFFFFFF
    network = ip_int & mask
    broadcast = network | (~mask & 0xFFFFFFFF)
    def to_ip(n): return '.'.join(str((n >> s) & 0xFF) for s in (24,16,8,0))
    return {
        'network': to_ip(network),
        'broadcast': to_ip(broadcast),
        'first_host': to_ip(network + 1),
        'last_host': to_ip(broadcast - 1),
        'usable_hosts': max(0, broadcast - network - 1)
    }

Expect these follow-ups

• How would you extend this to check whether two CIDRs overlap?
• What changes for a /31 network used on a point-to-point link?

As asked

The Cisco Meraki API enforces a rate limit of 10 requests per second per organization. Write a thread-safe token bucket rate limiter class in Python that callers can use before each API request. The bucket should refill at 10 tokens per second and support a configurable burst capacity.

Sample answer outline

A strong answer stores the token count and the last refill timestamp, calculates elapsed time on each acquire call to add tokens (capped at the burst capacity), and either returns immediately if a token is available or sleeps for the fractional second needed to refill one token. Thread safety requires a threading.Lock around the read-modify-write of both the count and the timestamp. The candidate should mention that this approach is slightly different from a leaky bucket and discuss the tradeoff: token bucket allows bursts up to the capacity, leaky bucket smooths traffic to an exact rate.

Reference implementation (python)

import time, threading

class TokenBucket:
    def __init__(self, rate: float, capacity: float):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self):
        with self.lock:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
            self.last = now
            if self.tokens >= 1:
                self.tokens -= 1
                return
            wait = (1 - self.tokens) / self.rate
        time.sleep(wait)
        self.acquire()

Expect these follow-ups

• How would you adapt this for an async context using asyncio.sleep?
• What happens if two threads both see tokens >= 1 before either decrements?

As asked

You need to collect data from three Cisco APIs in parallel: DNA Center device list, Meraki organization list, and Webex rooms list. Each requires its own authentication token. Write an async Python function that fetches all three concurrently, combines the results into a single dict, and handles the case where one API fails without canceling the other two.

Sample answer outline

A strong answer uses asyncio.gather with return_exceptions=True to run all three coroutines concurrently and collect results or exceptions without early cancellation. Each coroutine uses httpx.AsyncClient with its own auth header. The results list is inspected: entries that are exceptions get logged and excluded; valid responses get merged into the output dict. The candidate should mention that asyncio.gather does NOT cancel sibling tasks on exception when return_exceptions=True, which is the key behavior here.

Reference implementation (python)

import asyncio, httpx

async def fetch_all(dnac_token, meraki_key, webex_token):
    async def get_dnac():
        async with httpx.AsyncClient() as c:
            r = await c.get('https://dnac/dna/intent/api/v1/network-device',
                            headers={'X-Auth-Token': dnac_token})
            return ('dnac', r.json()['response'])

    async def get_meraki():
        async with httpx.AsyncClient() as c:
            r = await c.get('https://api.meraki.com/api/v1/organizations',
                            headers={'X-Cisco-Meraki-API-Key': meraki_key})
            return ('meraki', r.json())

    async def get_webex():
        async with httpx.AsyncClient() as c:
            r = await c.get('https://webexapis.com/v1/rooms',
                            headers={'Authorization': f'Bearer {webex_token}'})
            return ('webex', r.json()['items'])

    results = await asyncio.gather(get_dnac(), get_meraki(), get_webex(),
                                   return_exceptions=True)
    return {k: v for r in results if not isinstance(r, Exception) for k, v in [r]}

Expect these follow-ups

• What happens if one of these APIs has a 30-second timeout and the others complete in 2 seconds?
• How would you add retries to just the failing coroutine without retrying the ones that succeeded?

As asked

In Cloudflare Workers, implement a reusable middleware pattern that lets you chain multiple functions, each of which can modify the incoming Request or the outgoing Response, and can short-circuit the chain by returning early. Show how you would add authentication and logging as two separate middleware layers.

Sample answer outline

A strong answer composes middleware functions that accept (request, next) similar to Express or Koa, where next() calls the remaining chain. Authentication middleware checks the Authorization header and returns a 401 Response directly if invalid, bypassing next(). Logging middleware calls next(), awaits the Response, then logs both the request and the response status before returning. The candidate should handle errors in each middleware and understand that Workers' fetch event model is request-response, not streaming middleware.

Reference implementation (typescript)

type Middleware = (req: Request, next: () => Promise<Response>) => Promise<Response>;

function compose(...middlewares: Middleware[]) {
  return function (req: Request, finalHandler: () => Promise<Response>): Promise<Response> {
    let index = -1;
    function dispatch(i: number): Promise<Response> {
      if (i <= index) throw new Error('next() called multiple times');
      index = i;
      const fn = middlewares[i] ?? finalHandler;
      return fn(req, () => dispatch(i + 1));
    }
    return dispatch(0);
  };
}

Expect these follow-ups

• How would you pass context, like a user object extracted in the auth layer, to downstream middleware without mutating the Request?
• What happens to your middleware chain if the origin fetch times out?

As asked

Implement an in-memory cache with LRU eviction and per-key TTL in Python. It should support Get(key) returning None for missing or expired keys, Set(key, value, ttl_seconds), and run in O(1) for both operations. Expired keys should not appear in Get even if capacity is not exceeded.

Sample answer outline

A strong answer combines an OrderedDict (doubly linked list plus hash map) for LRU tracking with expiry timestamps stored alongside values. Get checks expiry first and treats expired entries as a miss without evicting eagerly (lazy expiry). Set inserts or updates, moves to tail, and enforces capacity. The candidate should discuss the tradeoff between lazy TTL cleanup (simpler, small memory leak) and background cleanup threads (lower memory, complexity).

Reference implementation (python)

from collections import OrderedDict
import time

class LRUCacheTTL:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.cache = OrderedDict()  # key -> (value, expire_at)

    def get(self, key: str):
        if key not in self.cache:
            return None
        value, expire_at = self.cache[key]
        if time.time() > expire_at:
            del self.cache[key]
            return None
        self.cache.move_to_end(key)
        return value

    def set(self, key: str, value, ttl_seconds: float):
        # TODO: implement
        pass

Expect these follow-ups

• How would you add a background cleanup thread that removes expired keys without blocking Get or Set?
• How does this design change if you need to support millions of keys across multiple Python processes?

As asked

You need to monitor a proprietary message queue that has no existing Datadog integration. It exposes an HTTP endpoint at /metrics that returns JSON with fields: queue_depth, consumer_count, messages_per_second, and oldest_message_age_seconds. Write a custom Datadog Agent check in Python that collects these metrics and emits them with the appropriate metric types and tags.

Sample answer outline

A strong answer extends the AgentCheck base class, implements the check(self, instance) method, uses self.http.get() or requests to call the endpoint (reading url from instance config), maps queue_depth to a gauge, messages_per_second to a rate or gauge, oldest_message_age_seconds to a gauge, and consumer_count to a gauge. The candidate should add error handling for a failed HTTP call (log and return, do not raise), tag with instance name and any other contextual tags from config, and show a minimal example conf.yaml.

Reference implementation (python)

from datadog_checks.base import AgentCheck

class CustomQueueCheck(AgentCheck):
    def check(self, instance):
        url = instance.get('url')
        tags = instance.get('tags', [])
        try:
            response = self.http.get(url)
            response.raise_for_status()
            data = response.json()
        except Exception as e:
            self.log.error('Failed to fetch metrics: %s', e)
            return

        self.gauge('queue.depth', data['queue_depth'], tags=tags)
        self.gauge('queue.consumer_count', data['consumer_count'], tags=tags)
        self.gauge('queue.oldest_message_age', data['oldest_message_age_seconds'], tags=tags)
        self.gauge('queue.messages_per_second', data['messages_per_second'], tags=tags)

Expect these follow-ups

• What is the difference between using self.gauge and self.rate for a counter metric?
• How would you test this check locally without a full Datadog Agent installation?

As asked

Write a function that takes a deeply nested JavaScript object and returns a flat object where each key is the dot-notation path to the original value. For example, { a: { b: { c: 1 } } } becomes { 'a.b.c': 1 }.

Sample answer outline

Use a recursive DFS or an iterative stack. For each key, if the value is a plain object (not array, not null), recurse with the accumulated path prefix. Otherwise write the value to the flat result. The candidate should handle arrays (either skip or treat indices as keys), null values, and non-enumerable keys.

Reference implementation (typescript)

function flatten(
  obj: Record<string, unknown>,
  prefix = ''
): Record<string, unknown> {
  // your implementation here
  return {};
}

// flatten({ a: { b: { c: 1 }, d: 2 } })
// => { 'a.b.c': 1, 'a.d': 2 }

Expect these follow-ups

• How would you write the inverse: unflattening 'a.b.c': 1 back to a nested object?