High-Level Architecture and System Design of PayPal’s JunoDB

JunoDB is PayPal’s open-source, highly available, distributed key-value store designed to handle massive scale with predictable low latencies. Built to power high-throughput workloads across global services, JunoDB achieves linear horizontal scalability and high availability through a decoupled, multi-tier architecture.

This article examines the core components of JunoDB, the design trade-offs behind its client-proxy-storage separation, and how consistent hashing and consensus engines like etcd are applied in practice.

---

1. High-Level Architectural Overview

JunoDB separates routing logic, configuration state, and raw storage into distinct layers:

+----------------------------------------------------------------+
|                           Clients                              |
|             (Go, Java, C++, Node.js SDKs)                      |
+----------------------------------------------------------------+
                                | (TCP)
                                v
+----------------------------------------------------------------+
|                        Load Balancer                           |
+----------------------------------------------------------------+
                                | 
        +-----------------------+-----------------------+
        |                                               |
        v                                               v
+------------------------------+       +------------------------------+
|         Juno Proxy           |  ...  |         Juno Proxy           |
| (Consistent Hashing Engine)  |       | (Consistent Hashing Engine)  |
+------------------------------+       +------------------------------+
        |                                               |
        +-----------------------+-----------------------+
                                | (Watches topology)
                                v
                     +--------------------+
                     |    etcd Cluster    |
                     +--------------------+
                                | (Persistent connections)
        +-----------------------+-----------------------+
        |                                               |
        v                                               v
+-------------------------------+       +-------------------------------+
|        Storage Server 1       |       |        Storage Server N       |
| +---------------------------+ |       | +---------------------------+ |
| | Shard 1 (RocksDB Instance)| |       | | Shard K (RocksDB Instance)| |
| +---------------------------+ |  ...  | +---------------------------+ |
| | Shard 2 (RocksDB Instance)| |       | | Shard M (RocksDB Instance)| |
| +---------------------------+ |       | +---------------------------+ |
+-------------------------------+       +-------------------------------+

---

2. The Storage Server: Sharding with Embedded RocksDB

The Storage Server is responsible for the actual persistence and retrieval of raw key-value pairs (supporting operations like GET, PUT, DELETE).

In-Memory vs. On-Disk Flexibility

JunoDB is not purely an in-memory cache; it supports durability. Depending on the workload and durability requirements, storage servers can be configured to hold data entirely in memory or persist records to NVMe/SATA disks.

Shards as Embedded RocksDB Instances

Rather than reinventing a custom local storage engine, JunoDB uses RocksDB—a high-performance, embedded LSM-tree key-value engine developed by Meta.

• Each storage server hosts multiple shards (partitions).
• Each shard is an independent, isolated RocksDB instance.
• Running multiple independent RocksDB instances per storage host limits blast radiuses, enables fine-grained concurrency, and simplifies shard migration between hosts.

Connection Handling

The storage server listens on a standard TCP port, accepting inbound operational requests from the proxy layer. When an incoming operation arrives, the server resolves which local RocksDB instance owns the key’s shard and commits the operation.

---

3. Shard Ownership & Consistent Hashing

To distribute keys evenly across nodes, JunoDB uses a two-step mapping process:

1. Key-to-Shard Mapping: Incoming keys are hashed and mapped to a fixed set of virtual shards (e.g., via a standard modulus or partition function).
2. Shard-to-Node Mapping (Consistent Hashing): The ownership of these shards across physical storage servers is governed by a Consistent Hashing Ring.

                   [Storage Node A] (owns Shards 1, 2)
                        /        \
                       /          \
                      /            \
    [Storage Node C]                [Storage Node B]
  (owns Shards 5, 6)                (owns Shards 3, 4)
                      \            /
                       \          /
                        \        /
                   [Storage Node D] (owns Shards 7, 8)

Why Consistent Hashing?

• Minimal Data Movement: When adding a new storage node or taking an existing node offline, only a small fraction ($1/N$) of shards need to be migrated to adjacent nodes on the ring.
• Deterministic Failover: If a storage server crashes, its assigned shards immediately map to the next consecutive active node on the hash ring without requiring a full cluster rebalance.

---

4. The Proxy Layer: Juno Proxy

A critical system design choice in distributed storage systems is choosing between direct client access and proxy-based access.

Feature	Direct-to-Storage Client	Proxy-Based Architecture (JunoDB)
Topology Awareness	Client SDK must track every storage host and ring change.	Storage topology is completely abstracted from the client.
Connection Overhead	High; $M$ clients $\times$ $N$ storage nodes creates connection explosion.	Minimal; Proxies multiplex client requests over persistent TCP pools.
SDK Maintenance	Complex logic must be ported across every programming language.	Client SDKs remain thin; routing and retry logic reside in proxies.
Security & Auditing	Difficult to monitor distributed ingress across all storage hosts.	Centralized control point for auth, metrics, and traffic shaping.

Multiplexing and Persistent Connections

Juno Proxy instances maintain warm, persistent TCP connection pools to all active storage servers. Client requests arriving at the proxy are rapidly matched against the consistent hash ring, routed to the target storage server, and returned.

---

5. Distributed Topology Coordination via etcd

Because JunoDB runs multiple stateless Juno Proxy instances behind a load balancer, every proxy must maintain an identical and up-to-date view of:

• The consistent hashing ring topology.
• Physical storage server health and availability.
• Shard-to-node ownership maps.

To maintain strong consistency across proxies, JunoDB uses etcd.

                     +-----------------------+
                     | Storage Cluster Event |
                     |  (Node Added/Removed) |
                     +-----------------------+
                                 |
                                 v
                     +-----------------------+
                     |   etcd RAFT Cluster   |
                     | (Source of Truth Map) |
                     +-----------------------+
                                 |
             +-------------------+-------------------+
             | (Watch Stream)                        | (Watch Stream)
             v                                       v
+-------------------------+             +-------------------------+
|      Juno Proxy 1       |             |      Juno Proxy 2       |
| [In-Memory Hash Ring]   |             | [In-Memory Hash Ring]   |
+-------------------------+             +-------------------------+

How etcd is Leveraged

1. Centralized Source of Truth: Any topology change (e.g., node commission, decommission, or failover) is committed to etcd using its Raft consensus protocol.
2. Low-Latency Watch Mechanism: Each Juno Proxy establishes a long-running watch on the configuration keys in etcd.
3. Atomic Ring Updates: When a change occurs, etcd immediately pushes the update to all proxies simultaneously, ensuring that all proxies update their local routing maps in near-lockstep.

---

6. End-to-End Request Lifecycle

Tracing a PUT or GET request from client execution to disk:

Client SDK       Load Balancer       Juno Proxy         Storage Server      RocksDB Shard
    |                  |                 |                    |                   |
    |--- 1. Request -->|                 |                    |                   |
    |                  |--- 2. Route --->|                    |                   |
    |                  |                 |-- 3. Hash Key &    |                   |
    |                  |                 |      Find Shard    |                   |
    |                  |                 |-- 4. Lookup Ring   |                   |
    |                  |                 |      in etcd Cache |                   |
    |                  |                 |                    |                   |
    |                  |                 |--- 5. TCP Forward->|                   |
    |                  |                 |                    |--- 6. Read/Write->|
    |                  |                 |                    |<-- 7. Result -----|
    |                  |                 |<-- 8. Response ----|                   |
    |                  |<-- 9. Response -|                    |                   |
    |<-- 10. Ack ------|                 |                    |                   |

1. SDK Request: The application initiates a request (e.g., client.Get("user_1234")).
2. Load Balancing: The request hits a standard layer-4 or layer-7 load balancer distributing traffic across available Juno Proxies.
3. Shard Resolution: The proxy hashes the key to determine its shard ID.
4. Host Resolution: The proxy inspects its local consistent hashing ring (continuously synced via etcd) to identify which storage server currently owns that shard.
5. Server Dispatch: The proxy reuses an existing persistent TCP connection to forward the command to the target storage server.
6. Storage Execution: The storage server identifies the local RocksDB instance bound to that shard ID and executes the read/write.
7. Return Pipeline: The result flows back through the storage server, proxy, and load balancer to the client SDK.

---

7. True Horizontal Scalability

Every layer in JunoDB scales independently according to demand:

• Load Balancer Tier: Scales using standard infrastructure techniques (DNS round-robin, Anycast, ECMP).
• Proxy Tier (Compute-Bound): Proxies are stateless. When network traffic, connection concurrency, or proxy CPU load increases, additional proxy instances can be launched behind the load balancer without data migration.
• Storage Tier (State-Bound): When disk capacity or storage I/O limits are approached, new storage servers can be added. The shards are reallocated along the consistent hashing ring, minimal data is copied between old and new nodes, and the new topology is broadcast to all proxies via etcd.

---

8. Key Takeaways

• Reusing Proven Components: Instead of building a storage engine from scratch, JunoDB utilizes RocksDB for single-node embedded storage and etcd for distributed configuration consensus.
• Decoupling Topology from Clients: Introducing a dedicated proxy layer prevents connection proliferation and keeps client SDKs lightweight.
• Predictable Data Placement: Combining key-to-shard partitioning with a consistent hashing ring ensures minimal data movement during scaling events and rapid, deterministic recovery from node failures.