Shoring Up Weak Points + a Coding Warm-Up: Rate Limiter, LRU, Concurrency Control, Streaming JSON Parsing

// Token bucket: capacity, refilled steadily at refillPerSec per second.
// Lazy refill - no timer; on take, refill by however long since the last refill.
// now is passed in by the caller (milliseconds) so tests can fast-forward without sleeping.
class TokenBucket {
  constructor(capacity, refillPerSec, nowMs) {
    this.capacity = capacity
    this.refillPerSec = refillPerSec
    this.tokens = capacity // a cold start is a full bucket: one burst of capacity is allowed
    this.lastRefillMs = nowMs
  }
 
  tryAcquire(cost, nowMs) {
    const elapsedMs = Math.max(0, nowMs - this.lastRefillMs) // a clock going back neither deducts nor gifts
    this.tokens = Math.min(this.capacity, this.tokens + (elapsedMs / 1000) * this.refillPerSec)
    this.lastRefillMs = nowMs
    if (this.tokens < cost) return false
    this.tokens -= cost
    return true
  }
}

import time
from dataclasses import dataclass, field
 
 
# Units are seconds, using time.monotonic's monotonic clock -
# unaffected by system clock changes, which beats basing it on time.time.
@dataclass
class TokenBucket:
    capacity: float
    refill_per_sec: float
    last_refill: float = field(default_factory=time.monotonic)
    tokens: float = field(init=False)
 
    def __post_init__(self) -> None:
        self.tokens = self.capacity  # a cold start is a full bucket
 
    def try_acquire(self, cost: float, now: float) -> bool:
        elapsed = max(0.0, now - self.last_refill)
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_per_sec)
        self.last_refill = now
        if self.tokens < cost:
            return False
        self.tokens -= cost
        return True

// synchronized is mandatory in the single-machine version: refill and deduct must sit in
// one critical section, or two threads both read "one left" and both pass.
// Units are milliseconds.
final class TokenBucket {
    private final double capacity;
    private final double refillPerSec;
    private double tokens;
    private long lastRefillMs;
 
    TokenBucket(double capacity, double refillPerSec, long nowMs) {
        this.capacity = capacity;
        this.refillPerSec = refillPerSec;
        this.tokens = capacity; // a cold start is a full bucket
        this.lastRefillMs = nowMs;
    }
 
    synchronized boolean tryAcquire(double cost, long nowMs) {
        long elapsedMs = Math.max(0L, nowMs - lastRefillMs);
        tokens = Math.min(capacity, tokens + elapsedMs / 1000.0 * refillPerSec);
        lastRefillMs = nowMs;
        if (tokens < cost) return false;
        tokens -= cost;
        return true;
    }
}

// Units are seconds. Wrap it in an actor or a lock when shared across tasks -
// refill plus deduct must be one indivisible action, which holds in all four languages.
final class TokenBucket {
    private let capacity: Double
    private let refillPerSec: Double
    private var tokens: Double
    private var lastRefill: Double
 
    init(capacity: Double, refillPerSec: Double, now: Double) {
        self.capacity = capacity
        self.refillPerSec = refillPerSec
        self.tokens = capacity // a cold start is a full bucket
        self.lastRefill = now
    }
 
    func tryAcquire(cost: Double, now: Double) -> Bool {
        let elapsed = max(0, now - lastRefill)
        tokens = min(capacity, tokens + elapsed * refillPerSec)
        lastRefill = now
        guard tokens >= cost else { return false }
        tokens -= cost
        return true
    }
}

// JS's Map preserves insertion order, so "promote" is delete then set and no list is needed.
class LRUCache {
  constructor(capacity) {
    if (capacity <= 0) throw new Error('capacity must be greater than 0')
    this.capacity = capacity
    this.map = new Map()
  }
 
  get(key) {
    if (!this.map.has(key)) return undefined
    const value = this.map.get(key)
    this.map.delete(key)
    this.map.set(key, value) // reinserted at the tail = most recently used
    return value
  }
 
  put(key, value) {
    if (this.map.has(key)) this.map.delete(key) // updating an existing key also refreshes order
    this.map.set(key, value)
    if (this.map.size > this.capacity) {
      const oldest = this.map.keys().next().value // the head = least recently used
      this.map.delete(oldest)
    }
  }
}

from collections import OrderedDict
 
 
class LRUCache:
    """Python has OrderedDict.move_to_end, half the length of a hand-written list.
    For a real business cache use functools.lru_cache; writing it by hand is because
    an interview wants to see the eviction order."""
 
    def __init__(self, capacity: int) -> None:
        if capacity <= 0:
            raise ValueError("capacity must be greater than 0")
        self.capacity = capacity
        self.store: OrderedDict[str, int] = OrderedDict()
 
    def get(self, key: str) -> int | None:
        if key not in self.store:
            return None
        self.store.move_to_end(key)  # moved to the tail = most recently used
        return self.store[key]
 
    def put(self, key: str, value: int) -> None:
        if key in self.store:
            self.store.move_to_end(key)  # updating an existing key also refreshes order
        self.store[key] = value
        if len(self.store) > self.capacity:
            self.store.popitem(last=False)  # pop the head = least recently used

import java.util.LinkedHashMap;
import java.util.Map;
 
// The JDK has this built in: accessOrder true gives LRU order, and removeEldestEntry is
// called back after each insertion, evicting the head when it returns true.
final class LRUCache<K, V> extends LinkedHashMap<K, V> {
    private final int capacity;
 
    LRUCache(int capacity) {
        super(16, 0.75f, true); // true = access order, so get also moves an entry to the tail
        if (capacity <= 0) throw new IllegalArgumentException("capacity must be greater than 0");
        this.capacity = capacity;
    }
 
    @Override
    protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
        return size() > capacity;
    }
}

// Swift's standard library has no order-preserving dictionary, so it is a hash map plus a
// doubly linked list by hand. Which makes this the closest to what an interviewer wants
// when they say "no built-in ordered containers."
final class LRUCache {
    private final class Node {
        let key: String
        var value: Int
        var next: Node?
        weak var prev: Node? // prev must be weak, or mutual references form a cycle and never free
        init(key: String, value: Int) {
            self.key = key
            self.value = value
        }
    }
 
    private let capacity: Int
    private var map: [String: Node] = [:]
    private let head = Node(key: "", value: 0) // sentinels, sparing half the nil checks
    private let tail = Node(key: "", value: 0)
 
    init(capacity: Int) {
        precondition(capacity > 0, "capacity must be greater than 0")
        self.capacity = capacity
        head.next = tail
        tail.prev = head
    }
 
    func get(_ key: String) -> Int? {
        guard let node = map[key] else { return nil }
        moveToTail(node) // a get counts as a use
        return node.value
    }
 
    func put(_ key: String, _ value: Int) {
        if let node = map[key] {
            node.value = value
            moveToTail(node)
            return
        }
        let node = Node(key: key, value: value)
        map[key] = node
        insertBeforeTail(node)
        // Insert before evicting; reversed, a capacity of 1 evicts what you just added
        if map.count > capacity, let oldest = head.next, oldest !== tail {
            unlink(oldest)
            map.removeValue(forKey: oldest.key)
        }
    }
 
    private func unlink(_ node: Node) {
        node.prev?.next = node.next
        node.next?.prev = node.prev
        node.next = nil
        node.prev = nil
    }
 
    private func insertBeforeTail(_ node: Node) {
        let last = tail.prev
        last?.next = node
        node.prev = last
        node.next = tail
        tail.prev = node
    }
 
    private func moveToTail(_ node: Node) {
        unlink(node)
        insertBeforeTail(node)
    }
}

// Limit simultaneously running async tasks to limit.
async function mapWithLimit(items, limit, worker) {
  const results = new Array(items.length)
  let cursor = 0
 
  // One runner is one concurrency slot, and the slot count is fixed at limit
  async function runner() {
    while (cursor < items.length) {
      const index = cursor++
      try {
        results[index] = { ok: true, value: await worker(items[index]) }
      } catch (error) {
        // A failure must return its slot: without catching here the runner exits and
        // that slot stays empty forever, quietly dropping concurrency below limit.
        results[index] = { ok: false, reason: String(error) }
      }
    }
  }
 
  await Promise.all(Array.from({ length: Math.min(limit, items.length) }, runner))
  return results
}

import asyncio
from typing import Awaitable, Callable, TypeVar
 
T = TypeVar("T")
 
 
async def map_with_limit(
    items: list[T], limit: int, worker: Callable[[T], Awaitable[object]]
) -> list[dict[str, object]]:
    sem = asyncio.Semaphore(limit)
 
    async def run(item: T) -> dict[str, object]:
        # async with always releases on leaving scope, equivalent to try/finally, so a
        # raising task cannot leak a slot - the most reassuring part of the Python version.
        async with sem:
            try:
                return {"ok": True, "value": await worker(item)}
            except Exception as exc:
                return {"ok": False, "reason": str(exc)}
 
    # gather only collects results; the semaphore above controls concurrency, not gather
    return list(await asyncio.gather(*(run(item) for item in items)))

// Dependencies: java.util.concurrent (JDK 17+)
// acquire before submitting: the main thread is blocked first, so the task queue cannot
// grow unbounded, which is natural backpressure.
static <T, R> List<String> mapWithLimit(List<T> items, int limit, Function<T, R> worker)
        throws InterruptedException {
    var sem = new Semaphore(limit);
    var pool = Executors.newCachedThreadPool();
    var futures = new ArrayList<CompletableFuture<String>>();
    try {
        for (T item : items) {
            sem.acquire();
            futures.add(CompletableFuture.supplyAsync(() -> {
                try {
                    return "ok:" + worker.apply(item);
                } catch (RuntimeException e) {
                    return "err:" + e.getMessage();
                } finally {
                    sem.release(); // returned on success and failure; this line is the crux
                }
            }, pool));
        }
        return futures.stream().map(CompletableFuture::join).toList();
    } finally {
        pool.shutdown();
    }
}

// withTaskGroup runs everything added via addTask concurrently and limits nothing itself.
// Concurrency needs your own window: fill limit slots, then add one per result received.
func mapWithLimit<T: Sendable>(
    _ items: [T],
    limit: Int,
    worker: @escaping @Sendable (T) async throws -> String
) async -> [String] {
    var results = [String](repeating: "", count: items.count)
    await withTaskGroup(of: (Int, String).self) { group in
        var next = 0
        func submit(_ index: Int) {
            let item = items[index]
            group.addTask {
                // Errors must collapse into result values inside the task: letting one
                // escape cancels the whole group, losing the remaining tasks' slots and results.
                do { return (index, "ok:" + (try await worker(item))) } catch {
                    return (index, "err:\(error)")
                }
            }
        }
        while next < min(limit, items.count) {
            submit(next)
            next += 1
        }
        for await (index, value) in group {
            results[index] = value
            if next < items.count {
                submit(next)
                next += 1
            }
        }
    }
    return results
}

// (a) The SSE kind, one complete JSON per line: line buffering plus per-line parse.
function createLineParser(onEvent) {
  let buffer = ''
  return function feed(chunk) {
    buffer += chunk
    const lines = buffer.split('\n')
    buffer = lines.pop() ?? '' // the tail may be a half line, kept for the next round (D1's trap)
    for (const line of lines) {
      const trimmed = line.trim()
      if (!trimmed.startsWith('data:')) continue // skip blank, comment, and event lines
      const payload = trimmed.slice(5).trim()
      if (payload === '[DONE]') return // it is not JSON, and parsing it definitely throws
      onEvent(JSON.parse(payload))
    }
  }
}

import json
from typing import Any, Callable
 
 
def make_line_parser(on_event: Callable[[dict[str, Any]], None]) -> Callable[[str], None]:
    buffer = ""
 
    def feed(chunk: str) -> None:
        nonlocal buffer
        buffer += chunk
        # Star unpacking in one step: the earlier items are complete lines and the last
        # is a half line kept for the next round
        *lines, buffer = buffer.split("\n")
        for line in lines:
            line = line.strip()
            if not line.startswith("data:"):
                continue
            payload = line[5:].strip()
            if payload == "[DONE]":
                return
            on_event(json.loads(payload))
 
    return feed

// Dependencies: Jackson (com.fasterxml.jackson.databind)
// Java has no built-in "split on a delimiter keeping the remainder", so buffer by hand.
final class LineParser {
    private final StringBuilder buffer = new StringBuilder();
    private final ObjectMapper mapper = new ObjectMapper();
    private final Consumer<JsonNode> onEvent;
 
    LineParser(Consumer<JsonNode> onEvent) {
        this.onEvent = onEvent;
    }
 
    void feed(String chunk) throws JsonProcessingException {
        buffer.append(chunk);
        int nl;
        while ((nl = buffer.indexOf("\n")) >= 0) {
            String line = buffer.substring(0, nl).strip();
            buffer.delete(0, nl + 1); // the remaining half line stays for the next round
            if (!line.startsWith("data:")) continue;
            String payload = line.substring(5).strip();
            if (payload.equals("[DONE]")) return;
            onEvent.accept(mapper.readTree(payload));
        }
    }
}

// In real code URLSession's bytes(for:).lines already does the framing and half-line
// stitching (D1 covered it); this hand-written version is for the whiteboard.
// Swift has Codable, so decode straight into a concrete type.
struct StreamEvent: Decodable {
    let delta: String
}
 
final class LineParser {
    private var buffer = ""
    private let decoder = JSONDecoder()
    private let onEvent: (StreamEvent) -> Void
 
    init(onEvent: @escaping (StreamEvent) -> Void) {
        self.onEvent = onEvent
    }
 
    func feed(_ chunk: String) {
        buffer += chunk
        var parts = buffer.components(separatedBy: "\n")
        buffer = parts.removeLast() // the tail may be a half line, kept for the next round
        for line in parts {
            let trimmed = line.trimmingCharacters(in: .whitespaces)
            guard trimmed.hasPrefix("data:") else { continue }
            let payload = trimmed.dropFirst(5).trimmingCharacters(in: .whitespaces)
            guard payload != "[DONE]" else { return }
            guard let data = payload.data(using: .utf8),
                  let event = try? decoder.decode(StreamEvent.self, from: data)
            else { continue }
            onEvent(event)
        }
    }
}