Design Your Own NoSQL Store — build it, don't read it

Something like: "A partition-tolerant, horizontally-scalable store for high-write-volume, key-addressable financial event data, where a single record's latest state must always be reconstructable even if a node holding it disappears."Why this framing: it commits to three things immediately — write-heavy, key-addressable (not relational), and durability-over-features — which rules out a large class of "nice to have" requirements before you even list them.

1. Accept a write for any key without requiring a schema migration.
2. Guarantee a write, once acknowledged, survives the loss of any single node.
3. Serve point reads by key with a bounded, predictable latency envelope.
4. Serve ordered range reads within one partition (e.g. one customer's transactions by time) without a full scan.
5. Support at least one secondary lookup path (e.g. by ISIN or email) with an explicitly stated staleness bound.
6. Allow the cluster to grow or shrink without a maintenance-window write pause.
7. Expose, per-write, a caller-chosen consistency level rather than one fixed global setting.

Why bullet 7 matters most: it's the one FR that prevents every future team from silently inheriting the wrong consistency guarantee for their data — which is exactly the incident pattern that shows up in mature systems.

Why the burst number matters more than the steady-state one: a store sized only for average load is precisely the store that falls over on the one day — an NFO open, an allotment run — where the business actually needs it most. NFR numbers that ignore your worst known day aren't targets, they're wishes.

Scope out: ad-hoc multi-key ACID transactions across partitions, a general-purpose query language, and strict global ordering across unrelated keys.Why: each of those three, if included, forces a coordination mechanism (distributed locking, a query planner, a global sequencer) that fights every partitioning and replication decision you're about to make in segment 2. A NoSQL store that tries to be "SQL but distributed" usually ends up being neither fast nor simple — scoping these out up front is what keeps the rest of this design honest.

Consistent hashing with virtual nodes for the partition key (e.g. customer_id or scheme_id), with data clustered/sorted by a secondary key (e.g. timestamp) within each partition.Why: pure hash partitioning gives you even spread and defuses the NFO-day hotspot from FR3/NFR-burst, while clustering within the partition still gives you the ordered range read from FR4 — you get both properties by applying them at two different levels of the key, not by picking one partitioning strategy and hoping it does both jobs.

Leaderless, quorum-based replication (any of the N replicas for a partition can coordinate a write; success requires W acks).Why: leader-based replication makes a leader-election pause a direct hit against your write-availability NFR every time a leader dies. Leaderless removes that single point of write unavailability at the cost of pushing conflict handling onto quorum math and read repair — a cost you're explicitly willing to take on, per FR2 and FR7, because durability-under-node-loss was stated as non-negotiable.

The consistency level per call — effectively, the caller's chosen W (write quorum) and R (read quorum) against your fixed replication factor N.Why: FR7 explicitly required per-call consistency rather than one global setting. Leaderless replication is the only topology of the two where that requirement is even structurally possible — a leader-based system's consistency story is baked into the topology itself, not exposed as a per-call dial.

Because they have opposite performance and failure profiles: the coordinator must decide "which replicas, what quorum, route now" in microseconds and must be stateless enough to fail over instantly, while the storage engine must guarantee a fsync-durable write even if that takes longer under load. Merging them means a slow disk on one node now blocks routing decisions for every key that node happens to coordinate — a local disk problem becomes a cluster-wide routing problem.

Because it has a fundamentally different lifecycle: Replication/Quorum Manager runs synchronously, in the write's critical path; Hinted Handoff runs asynchronously, potentially for hours, buffering writes meant for a down replica until it returns. Coupling them means a slow or full hint buffer can back-pressure onto live writes it has nothing to do with.

Because Gossip's job is to converge eventually across the whole cluster, while the Coordinator needs a fast, local, possibly-slightly-stale read of that state to route a single request in microseconds. If routing had to wait for gossip convergence on every call, you'd trade routing latency for membership freshness — the Coordinator instead reads a locally-cached view that Gossip updates in the background.

Because two APIs can be "owned" by the same team and still have wildly different blast radii — PUT /kv/{key} can only ever corrupt one key, while a leaked credential for /internal/replicate could let a caller impersonate a coordinator and write inconsistent data cluster-wide. Ownership tells you who to page; blast radius tells you how tightly to lock the door — and those are frequently different answers for APIs sitting right next to each other in the same service.

Because last-write-wins conflict resolution in a leaderless system needs a timestamp that reflects when the write was decided, not when it happened to arrive at whichever replica processed it — server-received time would make conflict resolution dependent on network jitter and routing luck rather than actual write order.Why this is a real risk, not a theoretical one: if two concurrent writes to the same key take different network paths, server-received time could resolve the conflict backwards from what the caller intended — client_timestamp (with a documented clock-skew tolerance) keeps that decision anchored to intent.

Table: mf_ledger_entries
Partition key:   customer_id (hashed)
Clustering key:  txn_timestamp DESC
Columns:         txn_id, scheme_id, units, nav, amount, status, vector_clock

Row size (est.):        ~250 bytes/row
Rows/customer/year:     ~60 (avg active investor)
Active customers:       ~4M
Rows/year:               240M
Storage/year (RF=3):     240M x 250B x 3  ~ 180 GB
Sustained write rate:    ~15K rows/sec platform-wide (matches NFR from segment 1)

Why the clustering key is txn_timestamp DESC, not just timestamp: the dominant read pattern is "most recent transactions first" for a customer's statement view — sorting descending at the storage layer means that query is a sequential prefix scan, not a full-partition scan followed by an application-side sort.

Durable config: the ring/partition map, replication factor, schema/column definitions. Ephemeral/derived state: the memtable contents, the read cache, gossip's locally-cached membership view — all rebuildable from the WAL or from peers.Why conflating them is a mistake: if you replicate ephemeral state (like a read cache) with the same durability guarantees as durable config, you're paying real replication cost to protect data that was never supposed to need protecting — it should have been rebuilt cheaply from source instead. The inverse mistake — treating durable config as rebuildable — is worse: lose the ring topology without a durable backup, and you don't know which node owns which data anymore.

At 15K writes/sec sustained with 180GB/year growth, a single powerful relational instance with read replicas could plausibly handle this — the honest answer is: the burst number (80K/sec on NFO day) and the multi-region durability requirement are what actually force the distributed design, not the steady-state volume.Why this question matters even after you've built the whole design: it's the question a sharp director asks at the review — "did we need this complexity, or did we default to it because it's the modern answer?" Being able to point at the specific number that justified it is the difference between an engineering decision and a trend follow.

struct Ring:
    sorted_positions: list<(hash_value, virtual_node_id, physical_node)>

function get_replicas(key, N):
    h = hash(key)
    idx = binary_search_first_position_gte(ring.sorted_positions, h)
    replicas = []
    physical_seen = set()
    while len(replicas) < N:
        (pos, vnode, node) = ring.sorted_positions[idx]
        if node not in physical_seen:
            replicas.append(node)
            physical_seen.add(node)
        idx = (idx + 1) mod len(ring.sorted_positions)
    return replicas

Why virtual nodes and physical_seen dedup both matter: virtual nodes spread each physical node's ownership across many ring positions, smoothing load; the physical_seen check prevents two virtual nodes belonging to the same physical machine from both counting toward your N — otherwise a "replicated" write could silently have fewer real physical copies than N implies.

function coordinate_write(key, value, W, N):
    replicas = get_replicas(key, N)
    write_ts = current_client_timestamp()
    acked = 0
    hinted = []

    for node in replicas:                     # fan out concurrently, not sequentially
        try:
            send_replicate(node, key, value, write_ts, timeout=WRITE_TIMEOUT)
            acked += 1
        except Timeout or NodeDown:
            hinted.append(node)                # buffer via hinted handoff, don't block

    if acked >= W:
        return SUCCESS
    else:
        # sloppy quorum: write to next available healthy node in the ring
        # instead of the "correct" but unreachable one, plus a hint for the real owner
        fallback_node = next_healthy_node_not_in(replicas)
        send_replicate(fallback_node, key, value, write_ts)
        acked += 1
        if acked >= W:
            return SUCCESS
        else:
            return FAIL_INSUFFICIENT_QUORUM   # client must decide: retry, or accept the write is not guaranteed

Why fan out concurrently, not sequentially: sequential replication would multiply write latency by N; concurrent fan-out means write latency is bounded by the slowest of the first W replicas to respond, not the sum of all N.Why sloppy quorum exists at all: without it, a single temporarily-down replica in a small ring could make writes to its owned keys fail outright even though the cluster has plenty of healthy capacity elsewhere — sloppy quorum trades strict ownership for availability, exactly the tradeoff FR2/NFR-availability asked for.

It doesn't need a lock, because each write is independently routed to the same N replicas and each replica resolves the conflict locally by comparing incoming timestamp against its currently stored value — last-write-wins, applied per-replica, per-key, on arrival.Why this avoids the coordination bottleneck a global lock would introduce: a lock would force every write to the same key through a single serialization point — exactly the kind of hotspot segment 3's component split was designed to avoid. Pushing conflict resolution to the replica, at apply-time, means concurrent writes to the same key are handled in parallel across replicas and simply converge to the same final answer via the same deterministic rule everywhere.

every REPAIR_INTERVAL (e.g. 24h), per partition range, per node:
    for each peer replica of this range:
        local_tree  = build_merkle_tree(local_range_data)
        peer_tree   = request_merkle_tree(peer, range)
        diff_leaves = compare_trees(local_tree, peer_tree)   # only mismatched sub-ranges
        for leaf in diff_leaves:
            local_data  = read_range(leaf.range)
            peer_data   = request_range(peer, leaf.range)
            merged      = resolve_by_timestamp(local_data, peer_data)  # same LWW rule as writes
            apply_missing_or_newer(local, merged)
            stream_missing_or_newer(peer, merged)

Why compare tree hashes before comparing any actual data: full record-by-record comparison across a large partition range is prohibitively expensive in bandwidth. Comparing small hash summaries first, and only drilling into the specific sub-ranges that differ, turns an O(all data) operation into something close to O(actual drift) — most of the tree usually matches and gets skipped entirely.

Distributed — each node schedules its own repairs against its peer replicas, staggered to avoid overlapping load.Why: a centralized repair coordinator becomes exactly the kind of single point of failure this whole design has been avoiding since segment 2 — if it goes down, repair silently stops cluster-wide with no local fallback. Distributed scheduling costs you a bit of coordination complexity (staggering to avoid every node repairing at once and saturating the network) but means no single component's failure can silently disable the entire cluster's self-healing.

Because using a different conflict rule for repair than for live writes would mean the "corrected" state after repair could disagree with what a live quorum read would have returned for the same conflicting writes — two different subsystems producing two different answers to the same question. Reusing one deterministic rule everywhere is what makes "eventually consistent" mean something concrete: eventually, every path converges to the identical answer, not just an answer.

            gossip missed heartbeats
HEALTHY  ------------------------->  SUSPECTED
   ^                                     |  \
   |            heartbeat resumes        |   \ SUSPECT_TIMEOUT elapses
   +-------------------------------------+    v
                                            DOWN
                                              |
                                   node process restarts / rejoins
                                              v
                                        REJOINING
                                     (replaying hinted handoff,
                                      catching up via repair)
                                              |
                                   caught up within threshold
                                              v
                                          HEALTHY

  any state --(operator triggers decommission)--> DRAINING --(data streamed off)--> REMOVED

Why SUSPECTED is a separate state from DOWN, not the same thing: a single missed heartbeat is often a transient network blip, not a real failure. Treating it as DOWN immediately would trigger unnecessary hinted handoff and repair traffic for every minor GC pause or network hiccup — SUSPECTED is a probation period that lets the node prove it's still alive before the cluster pays the cost of treating it as gone.

Two places: SUSPECTED could theoretically sit forever without a bound, and REJOINING could hang forever if hint replay or repair never actually finishes catching the node up.The fix: SUSPECTED has an explicit SUSPECT_TIMEOUT that forces a transition to DOWN, guaranteeing the cluster always eventually treats an unresponsive node as gone rather than leaving it in limbo indefinitely. REJOINING has an analogous catch-up deadline — if a node can't finish replaying hints and repairing within a bounded window, it's demoted back to DOWN rather than serving reads from a partially-caught-up state indefinitely, which would silently violate your consistency guarantees.

At the coordinator's FAIL_INSUFFICIENT_QUORUM outcome from segment 6 — if a write returns that failure because replicas were SUSPECTED or DOWN at the time, the client's retry is safe to send again because every write carries the same client_timestamp and last-write-wins semantics from segment 4. A retried write either lands as the first successful attempt, or loses harmlessly to whichever attempt's timestamp actually won — retries are safe precisely because the conflict-resolution rule was made deterministic back in segment 6, not because of anything new in the state machine itself.

Because convergence time for a rumor to reach every node scales roughly logarithmically with cluster size (each round roughly doubles the informed population) — cheap per-node, but the total time to full convergence still grows as the cluster grows, and it becomes a real problem once membership-change events (a node joining, dying, or being suspected) start happening faster than the cluster can converge on the previous change.Why this eventually caps how large one cluster should get: past a few hundred to low thousands of nodes, membership churn can outpace convergence, and the cluster spends more time disagreeing about its own topology than serving requests — the practical fix is usually sharding into multiple independently-gossiping clusters rather than growing one cluster indefinitely.

Because rebalancing means streaming partition data to the new node's virtual positions while production traffic keeps flowing — that streaming competes for the same disk and network I/O the live write and repair paths need, temporarily worsening the exact latency metric that triggered the scale-out decision.Why this has to be planned, not automatic: rebalancing throughput should be rate-limited below whatever headroom currently exists, the same way segment 6's write path avoided a global lock — trading a longer rebalance window for a smaller live-traffic impact, rather than rebalancing as fast as possible and risking the exact latency spike you were trying to fix.

The Coordinator never blocks a live request on Gossip convergence — it always routes off a locally-cached view of cluster membership, even though that view might be a few gossip-rounds stale.Why this was a deliberate choice: it directly implements the segment-1 decision to favor write availability over perfect real-time consistency of cluster metadata — a slightly stale ring view causes at worst a sloppy-quorum fallback (segment 6), which was already designed to handle exactly this case. Making the hot path wait for control-path convergence would trade a rare, already-handled edge case for a routine tax on every single request.

This one is genuinely yours to find — but the shape of the exercise is: pick each FR and NFR line from segment 1 one at a time, and point to the exact component and mechanism in your diagram that satisfies it. If you land on any requirement where you can't point to a specific box — a secondary-index staleness bound with no component actually enforcing it, an availability target with no fallback path that actually kicks in — that's not a failure of this session. That gap is the most valuable thing this exercise produces: a specific, named hole in your own design, found by you, before a staff engineer or a production incident finds it for you.