There is something humbling about trying to build a simplified version of Gmail and Google Drive and realizing just how much engineering goes into making something that "just works." That was the core lesson from PennCloud, a semester-long project for CIS 5050: Software Systems at Penn where my team built a distributed cloud platform entirely from scratch in C++. No external databases, no cloud infrastructure, no third-party key-value stores. Just raw sockets, threads, and a lot of debugging at 2am.
This is the story of how we built it, what broke, and what we learned.
What we actually built
PennCloud is a cloud platform with two core services: a webmail service similar to Gmail and a file storage service similar to Google Drive. Users can sign up, log in, send and receive emails (including to real external addresses like Penn and Gmail accounts), upload files of any type up to 800MB, organize them into folders, and do all the standard operations you would expect: rename, move, delete.
Behind the scenes, five backend servers store all the data in a distributed key-value store. Five frontend servers handle HTTP and SMTP traffic. A load balancer sits in front of the frontends. A master coordinator manages which backend server is responsible for what. And an admin console lets you kill and restart individual servers to test fault tolerance.
The entire thing is written in C++, speaks raw HTTP/1.1 and SMTP, and runs on Linux. No frameworks, no shortcuts.
PennCloud system architecture, end to end.
The foundation: a distributed key-value store
Everything in PennCloud, user accounts, emails, files, session data, lives in a distributed key-value store that we built ourselves. Think of it as a giant sparse table: rows identified by a string key, columns identified by another string key, and binary values that can hold anything from a password hash to the raw bytes of a PDF.
The store supports four operations: PUT, GET, CPUT (conditional put, which only writes if the current value matches an expected value), and DELETE. CPUT is the interesting one. It is how you handle concurrent updates safely without a traditional transaction system. When a user sends an email, for example, we need to atomically add it to their inbox list. We read the current list, append the new entry, and then CPUT back: "write this new list only if the list has not changed since I read it." If someone else updated it in the meantime, we retry. It is a lightweight form of optimistic concurrency control.
To distribute the data across servers, we partition it into 10 tablets based on the first character of the row key. Row keys starting with a or b go to tablet 0, c or d go to tablet 1, and so on through w-z. Each tablet is stored on exactly 3 of the 5 backend servers, so there is always redundancy.
Replication: making sure three servers agree
Having three copies of every tablet is only useful if they actually stay in sync. For this we used primary-based replication. For each tablet, one server is designated the primary. All writes go through the primary: it applies the change locally and then pushes it to the two backup replicas. If a backup is unreachable, the primary skips it and moves on. The backup will catch up later when it comes back online.
If a client sends a write to a non-primary server, that server does not reject it. It forwards the request to the primary on the client's behalf. This keeps the API simple: clients do not need to know which server is primary, they just need to know which servers hold a given tablet.
The master coordinator is responsible for tracking which server is currently primary for each tablet and broadcasting that information to all backends every 250 milliseconds via heartbeats. This means if a primary goes down and a new one is elected, every server in the cluster knows about it within one heartbeat cycle, roughly a quarter of a second.
The master: a coordinator that stays off the data path
The master is intentionally simple. It does not store any data. It does not handle any reads or writes. Its only job is to know who is alive, who is primary for each tablet, and to tell everyone else.
Backend servers send a heartbeat to the master every 250 milliseconds. The master replies with the current primary map: a compact list of tablet-to-primary assignments. If a server misses 4 consecutive heartbeats (1 second of silence), the master marks it as dead and immediately promotes the next alive replica as primary for any tablets that server owned.
Frontend servers cache the primary address for each tablet locally. On failure, they retry once, and if that fails too, they do a fresh LOOKUP to the master to get the updated address. This design keeps the master off the critical path for normal operations: the master is only contacted when something changes or a cache needs refreshing.
The hard part: crash recovery
This is where things get genuinely interesting, and honestly where I spent most of my time.
The problem is straightforward to state: if a backend server crashes mid-operation and loses all its in-memory state, how do you get it back to exactly where it was? You cannot just replay every write since the beginning of time. And you cannot afford to write everything to disk on every operation because that would be unbearably slow.
The answer is a combination of write-ahead logging and periodic checkpointing.
Write-ahead logging means that every write (PUT, CPUT, DELETE) is appended to a log file on disk and fsynced to physical storage before the in-memory data is updated. The fsync is the critical part. Without it, the OS might buffer the write in RAM and the file on disk could still be empty when the server crashes. With fsync, you have a guarantee: if the server acknowledged a write to a client, that write is on disk.
The ordering matters too. Write to disk first, then update memory. If you do it the other way around and crash between the two steps, you have a write that exists in memory but not on disk. When you restart, that write is silently lost. With WAL, the worst case is replaying a write that already happened, which is safe because writes are idempotent in our case.
Checkpointing solves the problem of unbounded log growth. Every 60 seconds, a background thread serializes the entire in-memory tablet data to a binary snapshot file on disk. Once the checkpoint is written, the corresponding log file is no longer needed and a fresh one is started. On recovery, instead of replaying the entire history from day one, you load the latest snapshot (which captures everything up to that point) and then replay only the log entries that came after it.
One subtle issue: writing a large checkpoint file is not atomic. If the server crashes halfway through writing it, you end up with a corrupted checkpoint. The fix is simple: write to a temporary file first, then atomically rename it to the final path. On POSIX systems, rename is guaranteed to be atomic. The final path always contains either the old complete checkpoint or the new complete one, never a partial write.
During checkpointing, we also pause all writes to the tablet using a read-write lock. Normal concurrent reads and writes take a read lock and can proceed simultaneously. The checkpoint thread takes a write lock, which blocks until all ongoing writes finish and then freezes new ones until the snapshot is done. This guarantees the checkpoint captures a consistent state rather than a mixture of before and after some concurrent write.
Remote recovery: catching up after a crash
Local recovery (load checkpoint, replay log) gets you back to where you were when you crashed. But if the server was down for a while, other servers kept accepting writes. You are now behind.
After local recovery, the server contacts the current primary for each of its tablets and says: "my sequence number is N, what have I missed?" The primary checks its own WAL and sends back all entries with a sequence number greater than N. The recovering server applies them to memory and appends them to its own log, and it is back in sync.
There is a complication though. If the server was down long enough that the primary took a new checkpoint and deleted the old log, the primary's current log does not go back far enough to cover what the recovering server missed. We detect this case: the primary returns zero entries but its sequence number is higher than ours. In that situation, the recovering server requests the primary's full latest checkpoint plus current log, writes them to its own disk, and re-runs local recovery from those files. This two-path design avoids unnecessary full transfers in the common case while still handling the edge case correctly.
The frontend: more than just a web server
The frontend servers are not just HTTP handlers. Each one also runs an SMTP server to accept incoming emails from external clients like Thunderbird. A request comes in, poll() detects it, and the server spawns a worker thread to handle it.
One interesting optimization is the KV client connection pool. Every time a frontend needs to talk to a backend, it needs a TCP connection. Opening a new connection for every operation is expensive. So the KV client maintains a pool of up to 32 idle TCP connections to backend servers, reusing them across requests. This made a noticeable difference in response times under load.
File storage uses a schema worth describing. Files are stored under a generated ID (a combination of timestamp, process ID, and a sequence number for uniqueness). The actual file bytes live at a "content:file_id" column. A separate "path:absolute_path" column maps the human-readable path to the file ID. Renaming or moving a file is therefore just updating the path column to point to the same file ID. No copying of actual data. A 500MB file rename takes the same time as a 1KB file rename. We were able to upload files up to 800MB and confirmed 100MB uploads in about 7 seconds.
What broke and what we learned from it
Two problems stand out from this project. Both of them took real time to figure out, and neither showed up in the lectures. They were the kind of bugs you only catch when you actually run the system and watch it behave badly.
The first was a subtle recovery bug. Local recovery worked. Remote recovery worked when servers were only down for a few seconds. The problem showed up when we started simulating longer outages.
The way our recovery protocol worked: when a server came back online, it asked the current primary "my last sequence number is N, send me everything after that." The primary read its WAL, pulled out the matching entries, and shipped them over. Simple enough.
But here is the edge case we missed initially. If the recovering server had been down for more than 60 seconds, the primary had taken at least one checkpoint in the meantime. And after taking a checkpoint, the primary rotates its WAL: the old log gets deleted, a fresh one is started. So now the primary's current WAL no longer contains the entries the recovering server is asking for. The primary truthfully responds with zero matching entries. The recovering server thinks it is up to date. It is not.
The fix was to detect this case explicitly. If the primary returns zero entries but its own sequence number is higher than the recovering server's, that mismatch is the signal that we have fallen too far behind to use incremental sync. The recovering server then switches strategies: it fetches the primary's entire latest checkpoint plus current WAL, writes both files to its own disk, and re-runs local recovery from scratch. It is essentially saying "forget what I had, just give me your current state."
What made this hard to find is that the bug only manifests at specific timing intersections. You need the server to be down for at least one full checkpoint cycle, and then you need writes to be flowing while it is down. In normal manual testing, you never see it. You only catch it when you intentionally simulate long outages and then start asking "wait, why does this server still have stale data?"
The second hard problem was getting all the servers to actually talk to each other reliably. When everything is local on one machine, you can be a little sloppy with sockets and it usually works. When you have five backends, five frontends, a master, and a load balancer all making cross-process requests, every shortcut comes back to bite you.
The first thing we got wrong was message framing. Initially we used newlines to delimit messages, which works fine until you start sending binary file content that happens to contain a newline byte. The receiver thinks the message ended early, parses garbage, and the connection breaks. We had a real incident where uploading a particular PDF would corrupt the parsing state of the backend it was sent to. The fix was to switch the entire protocol to length-prefixed framing: every message starts with a header specifying exactly how many bytes follow. The receiver reads precisely that many bytes and does not scan for any terminator. This made the protocol completely binary-safe and removed a whole category of bugs.
The second issue was connection management. Opening a fresh TCP connection for every operation between a frontend and a backend turned out to be brutally slow. Each three-way handshake takes a few milliseconds, and a single page load can trigger dozens of KV operations. The latency added up to something a user could perceive. The fix was a connection pool. Each frontend now maintains up to 32 idle TCP connections to backends and reuses them across requests. The improvement in response time was immediately visible the moment we deployed it.
The third issue was about how frontends discovered the current primary for a tablet. Asking the master on every single request would have made the master a bottleneck and defeated the whole point of having a separate coordinator. So we cache the primary address locally on each frontend. The frontend only contacts the master if its cached primary fails twice in a row, on the assumption that the primary has changed. This keeps the master almost entirely off the critical path for normal traffic. It only gets involved during failover, which is exactly when you want it involved.
What we would do differently
A few things stand out in hindsight.
Passwords are stored in plaintext in the KV store. We knew this during development and it was a deliberate shortcut given the time constraints, but hashing them would be the obvious first fix.
The replication protocol we used (primary pushes to backups synchronously before acking the client) gives you strong consistency but hurts write latency under load. A quorum-based approach where you write to 2 out of 3 replicas and ack immediately, then lazily sync the third, would be faster and still fault-tolerant.
Tablet splits are static. If all your users happen to have usernames starting with the letter a, one tablet is overloaded while the others sit idle. Dynamic tablet splitting based on load would be the production-grade solution.
Finally, frontend operations are not fully atomic. Uploading a file involves multiple KV operations: write the content, write the path mapping, update the file list index. If the server crashes between any of these steps, you can end up with orphaned data. Proper transactions across multiple KV rows would fix this cleanly.
Final thoughts
Building PennCloud was the most educational project I have done at Penn. Reading about distributed systems concepts like replication, fault tolerance, and crash recovery is one thing. Debugging why your recovery code loses two writes out of every thousand under a specific timing condition is another. The gap between understanding a concept and implementing it correctly is enormous, and closing that gap is what makes projects like this worthwhile.
If you want to go deeper on any of the topics covered here, particularly the recovery system or the KV store design, I am happy to write a follow-up. The recovery protocol especially has a lot of nuance that this post only scratches the surface of.