We used a database as a message queue. Now we use Kafka.
FoundationDB was our only database, queue included. We built Apple's QuiCK design on top of it, hit the write limits, and moved async tasks to Kafka.
Contents
FoundationDB is the only database we use. This should surprise you since FoundationDB is pretty barebones, just a key-value store. It stores everything for us: tenants, object metadata, the replication log for data distributed across regions, etc. We also use it as a queue to handle async tasks, à la QuiCK, the queuing system Apple uses for CloudKit. This has scaled very nicely. I’m not surprised; it’s the same tech behind iCloud, a platform with at least 900 million users. Furthermore, keeping the queue inside FoundationDB means all transactions stay in the database, eliminating the dual-write problem.
So why start using Kafka now?
We’ve seen a few issues using our database as a message queue:
- Scheduling requires many writes and scans, which puts read load on FoundationDB that directly competes with user requests.
- Each task is expensive and needs multiple writes to complete (enqueue, claim, lease, etc). We have ever more tasks as we add more features.
- New team members have to learn all the custom code resulting from actually implementing the QuiCK paper. There’s no standard implementation, even though it’s a well known pattern in theory. Finesse is not something you can learn from a paper.
This isn’t a story of a neat 1:1 replacement. We still have the queues in FoundationDB. We moved asynchronous tasks like garbage collection to Kafka, we can reduce the read and write load on FDB and shave off a good amount of that pesky custom code. Read more to see how it all turned out for us!
You might tell us we should have “just used Kafka” the whole time. Beyond the fact that you used the j-word: have you ever waited for your not-even-that-big broker to catch up on a cold start? Do you know what a zookeeper is and why you don’t pay to take care of the animals? The poor zookeeper can’t even pet them. Have you ever felt like a plastic bag drifting through the wind but unable to start again because of the sheer madness that comes with spending months permuting JVM flags to try to eke out a spectre’s worth of performance so that your servers aren’t constantly on fire?
No? Just me?
Either way we kinda wanted to avoid Kafka because running it yourself is the administrative experience of finding yourself turned into a monstrous vermin and everyone around you is mildly annoyed at your experience and asking you to move on with life instead of understanding that you can’t work anymore because your arms have turned into dozens of legs. By the way, that’s actually what people mean when they call something “kafkaesque”, not a Qu’vatlh of paperwork.
Beyond the naïve way to build a queue on FoundationDB
So you need a queue. The FoundationDB docs contain tutorials for making simple queues in multiple languages. At a high level you put messages in on one end of the keyspace (namespace for keys) and then read them out of the other end of the keyspace. This works fairly well (if you’ve ever used Sidekiq, this model should be very familiar), but the main problems come with naming the entries in the queue.
The naïve way to do it is to use the FoundationDB equivalent of MySQL’s
AUTO INCREMENT where you assign each queue item its own atomically increasing
integer ID, but what happens when you have more than one producer?
Given a sufficiently distributed system, it's easy for two jobs to have conflicting IDs, such as two tasks getting the ID 67 and conflicting with eachother on insert. Sure, with enough work you can random or UUID your way out of this, but the core problem is that consuming work deletes it from the database. If a worker dies while it's processing an item, there's no way for another worker to retry. Once the worker consumes a job, it's no longer in the queue and that job dies with it.
t1 t2 t3──┬─────────────────────┬─────────────────────┬──────────────────────────▶│ │ │└─ worker pops └─ row gone from └─ worker dies('q',1) FoundationDB mid-PutObject┌──────────────────────────────────────────────────────────────────┐│ nothing in the database remembers that ('q',1) existed. ││ the replication job for uploads/report.pdf is simply gone. │└──────────────────────────────────────────────────────────────────┘// a user in Frankfurt cannot see an object written in Chicago, and// there is no record left anywhere that tells you why.
We really don’t want to just lose queue items because this could mean that
running a PutObject to the Tigris region in Chicago makes the object not show
up as visible to a user that just happens to be in Europe and is hitting the
Frankfurt region. It sure would be convenient if someone had thought of this
problem in detail and published a detailed description of how they solved it.
Foreshadowing is a literary device in which the author creates dramatic tension from which to signal to the reader that something is about to happen.
We implemented an Apple paper that describes how to implement durable message queues on top of FoundationDB
Turns out Apple has thought about this and implemented it for CloudKit with a system they call QuiCK, or a Queueing System in CloudKit. Our industry has silly names for things. QuiCK is a robust queuing system that uses FoundationDB’s Record Layer to implement a message queue based on time. To understand why this is such a galaxy-brained genius move, let’s take a sidestep into how time works in distributed systems.
Much ado about time
FoundationDB is an ordered key-value store, and the ordering is a huge bit of how it’s (ab)used in practice. One of the nice things about time is that generally it’s an ordered phenomenon. These two match.
In most temporal reference frames, events happen sequentially:
time│t1 ├──▶ person picks up apple│t2 ├──▶ person eats apple▼// one clock, one observer. the order you saw is the order it// happened. distributed systems do not get this for free.
Now, that “generally” in the previous paragraph is a bit of a misnomer in distributed systems. It helps to think about every part of a distributed system having its own independent observation of time and that changes a lot about how event ordering can be strange in practice.
Imagine that you have two reference frames: one is closer to the person interacting with that apple and the other is farther away from it and gets news about the apple from the initial reference frame tweeting about it over UDP for some reason.
RF1 · next to the apple RF2 · reads the tweets┌──────────────────────────┐ ┌──────────────────────────┐│ t1 picks up apple │ ── tweet ──▶ │ t3 reads "eats apple" ││ t2 eats apple │ ── tweet ──▶ │ t4 reads "picks up" │└──────────────────────────┘ └──────────────────────────┘// the first tweet was slow. RF2 believes the apple was eaten and// then picked up. attach the observed time, or live with this.
Oh no! The first tweet was slow and the clocks are slightly out of sync, so the second reference frame saw the tweets out of order! The obvious answer here is to attach the observed time to each tweet (and maybe get better time synchronization in your clusters). This makes everything in the distributed system have at least some understanding of when things happen and in what order they should have been observed in.
Much a-queue about time
As such, the message queue Apple built on top of FoundationDB uses time as a key part of job identity. Jobs are identified by their task type, item space, execution / vesting time, priority, and a unique ID.
("queue", version, taskType, itemSpace, vestingMs, priority, id)│└──▶ QueueItem┌─────────────┬──────────────────────────────────────────────────────────┐│ vestingMs │ when the job may run. this is the sort key. │├─────────────┼──────────────────────────────────────────────────────────┤│ taskType │ which queue. gc, lifecycle, replication. ││ itemSpace │ the tenant or bucket the job belongs to ││ priority │ ordering inside one vesting millisecond ││ id │ random suffix, so two producers never collide │└─────────────┴──────────────────────────────────────────────────────────┘// identity is mostly a timestamp. that one choice is the whole trick.
This seems a bit excessive until you realize that queue workers will fail, crash, and die, but it’s unacceptable to lose work in the process like if you did it naïvely. If the user doesn’t get the email for their birthday but their friends do, that’s an angry tweet and negative feedback on Yelp (the last bastion of real human contact, for now).
As such, here’s what the entire lifecycle flow for the message queue looks like.
First, application code does something that needs to schedule eventual work to be done, such as a user uploading an object to Tigris. The queue item is constructed and put into the queue to be executed in the near future:
t1 t2──┬─────────────────────────┬────────────────────────────────────────────▶│ │└─ PutObject commits. └─ vestingMs reached.the same txn writes a range scan canthe gc job at t2 see it now// the object and the job land together or not at all. there is no// second system to keep in sync, so there is no dual write.
Every so often a worker will poll the queue for work by asking the queue for available keys from the beginning of time until now. The worker will then randomly select jobs it’s interested in and then push them into the future to claim them. Once the job is claimed, it starts doing whatever the job requires it to do.
w-7 wakes up and asks the queue for every job that hasalready vested: getRange( (gc, 0) .. (gc, now) )◀──────── the scan reaches this far ────────▶ the future┌─────────────────────────────────────────────┬────────────────────────────┐│ [job1 t1] [job2 t2] [job3 t5] │ [job4 t9] [job2 t8] │└─────────────────────────────────────────────┴────────────────────────────┘0 │ now = t6└─── clear at t2, set at t8, workerId = w-7 ───▶when now catches up to t8, job2 drops back into the window for anyone to take// there is no claim flag and no lock table. moving the key// out past now is the claim, and t8 is the deadline.
Then the job finishes and the worker responds by deleting it from the database. But if the worker crashes while processing the job for some reason, another worker will pick it up when it becomes eligible for processing again. Jobs are either done instantly or eventually.
┌──────────────────┐ ┌──────────────────┐ ┌────────────────────────┐│ vested │──▶ │ leased by w-7 │──▶ │ w-7 crashed │└───────┬──────────┘ └───────┬──────────┘ └──────────┬─────────────┘▲ │ finishes ││ ▼ ││ ┌──────────────────┐ ││ │ cleared │ ││ └──────────────────┘ │└───────── lease expires, job vests again ─────────┘// crashing does not remove a job. it puts it back on the wheel one// lease period later, for whichever worker shows up next.
So in the best case every job gets picked up once, gets processed, achieves enlightenment, and then re-enters the cosmic background radiation until its time is needed again. Otherwise jobs end up in an endless wheel of saṃsāra where they just bounce between workers until one of them doesn’t fail to process it (or if it’s pathologically crashing workers, then a new worker binary is pushed that won’t crash this time, we hope).
Workers will own no jobs yet will still be happy
One of the key parts of this is the idea of workers leasing jobs instead of claiming them like they would in other queue worker systems. When a worker leases a job, it marks itself as having its greasy paws on the job in the database and then pushes it forward so that it can be taken over if it crashes. In general, every worker has their own unique ID and every class of job has its own fixed lease time, so when workers crash it’ll take at most the lease time for the jobs to be picked up by another worker and for the spice to continue flowing.
Also when a worker takes longer than the lease time to get something done, it re-leases jobs so that other workers don’t pick them up and step on the worker currently in progress:
worker1 is diffing a whole bucket. the work outlasts one lease.t2 t8 t14 t20 t26──┬────────────┬────────────┬────────────┬────────────┬──────────────────▶│ │ │ │ │lease renew renew renew done▲ ▲ ▲ ▲vest = t8 vest = t14 vest = t20 vest = t26◀════════════▶no other worker can see job2 for this whole span// while a worker keeps checking in, nobody else can see the// job. stop checking in and it vests again, for any reason.
The thorns in the roses
From what I’ve seen in practice and what I can crack from the technical paper, this really does seem to get you most of the durability and reliability guarantees you’d expect from Kafka or other message queues without having to have a second system in the mix and its associated transactions. However there’s some thorns in the roses that are worth mentioning:
- Everything being in FoundationDB means you're inviting transaction conflicts as part of the core way your message queue works. This is a huge part of why the Apple paper has workers randomly claim jobs instead of being deterministic about it. The randomness doesn't eliminate conflicts, but it makes them less likely.
- The randomness also can result in "chronically unlucky" jobs that just never get picked. I'll expand on this some more later.
- You end up creating a lot of write pressure on your FoundationDB cluster once you reach a fairly large scale (eg: millions of items backlogged in the queue). This ends up being a problem when the storage server is not able to keep up with the logs. Again, I'll expand on this some more later in the post.
- You end up needing to make sure your clocks are synced. They should be already, but worst case you need to set up a stratum 1 NTP server or do GPS time synchronization.
We made some changes
I mentioned that Apple published a paper, not a GitHub repo or source tarball. As a result, this means we needed to adapt this design for our use case, and write all the code ourselves. In the process we changed some things:
- We don't use FoundationDB’s Record Layer, we use the corpse of our pre-pivot database product as the record/query layer for FoundationDB.
- Apple's system is designed to accommodate dynamic numbers of queues, we have a fixed number of queues. Removing that stipulation removed a lot of complexity.
- We ended up needing the concept of "long-running" jobs that checkpoint their status into the message queue so another worker picking them up doesn't have to start over from scratch.
- We also needed cronjobs. Time being a huge part of the job identity makes this way less complicated in practice.
The wrath of cron
One of the key advantages of a system like QuiCK is that there’s no central coordination layer, so you don’t have to maintain another service that has wide ranging downstream effects upon failure. One of the key disadvantages of a system like QuiCK is that you don’t have a central coordination layer so you end up needing to build things like recurring/cronjobs by yourself. The fun part about how this is implemented is that jobs self-identify by when they get executed, so the cheeky way to implement cronjobs is to just have a job that reschedules itself when it’s done executing. This is what we ended up doing.
the bucket sweep, enqueued by hand exactly once, at t0t0 t60 t120 t180 t240──┬───────────────┬───────────────┬───────────────┬───────────────┬──────▶│ ▲ ▲ ▲ ▲run ── books ───┘ │ │ │run ── books ───┘ │ │run ── books ───┘ │run ── books ───┘run ───▶// the last thing every run does is schedule the next one. no cron// service, no leader election, nothing extra to page you about.
When you add up the durability, temporal smear from workers occasionally retrying jobs, and other third order effects of this design, you end up with all of the benefits of something like cron or systemd timers without having to actually run a cron service in your cluster. You basically farm out the scheduling work to the workers themselves. This is kinda beautiful in a way that I’m having trouble describing succinctly, it’s great.
We sharded the message queue
Apple’s paper recommends having workers randomly pick jobs to ensure that they don’t step on eachother when claiming them in FoundationDB. In practice we found that explicitly sharding jobs by the cryptographic hash of their key was a lot more useful. One of the downsides of randomly picking jobs is that with a sufficiently small number of workers, it’s hard to find a random selection percentage that has low conflict rates and doesn’t let jobs lag behind realtime too much.
six workers, each reaching for a fifth of what it can see┌────────┬───────────────────────────────────────── ────────────────────────┐│ job │ t2 t8 t14 t20 t26 t32 waiting │├────────┼─────────────────────────────────────────────────────────────────┤│ b17c │ taken ││ c04e │ skip taken ││ a91f │ skip skip skip skip skip skip waited 0s │└────────┴─────────────────────────────────────────────────────────────────┘// the coin has no memory of how long a91f has been waiting, so// it can sit there for minutes while nothing is wrong.
In order to work around this, we explicitly sharded jobs into “lanes”. Every worker monitors a few lanes, and when jobs get claimed, their vesting time changes, which makes the key change, which makes the lane change. This means that jobs constantly bounce between lanes as different workers cycle in and out so that any one worker or any one “unlucky” lane is automatically mitigated against at the infrastructure level.
queue · its own keyspace. no worker keeps anything of its own.┌──────────────────────────────────────────────────────────────────────────┐│ job1 {"ck":"03"} job2 {"ck":null} job3 {"ck":null} job4 {"ck":null} │└───────────────┬─────────────────────────────────────────┬────────────────┘│ lease checkpoint ▲▼ │┌───────────────┴─────────────────────────────────────────┴────────────────┐│ lane 0 w-7 [░░░] w-9 [░░░] w-2 [░░░] │├──────────────────────────────────────────────────────────────────────────┤│ lane 1 w-9 [░░░] w-3 [░░░] w-5 [░░░] │├──────────────────────────────────────────────────────────────────────────┤│ lane 2 w-1 [░░░] w-4 [░░░] w-9 [░░░] │└──────────────────────────────────────────────────────────────────────────┘// hash(job key) picks the lane, and claiming a job changes the// key, so a job does not stay in one lane or on one worker.
This is another case of something that would normally be solved by some kind of central management system with other message queue systems, but by not having one and being a bit clever about the design you can completely mitigate that entire failure mode.
We ended up with Kafka anyways
But yeah, here's the part where we hit the limits of our setup and had to swallow our pride before setting up Kafka anyways.
FoundationDB is really bad at coping with large sustained amounts of inserts in
a single cluster. Sure, commits are fast, but at some point the storage servers
have to apply and replicate out those changes. When those storage servers fall
behind the transaction logs, the ratekeeper of the database throttles everyone
so they can catch up. As our customers start storing more and more data, we
started hitting this issue pretty regularly. Every PutObject call turned into
more and more writes which resulted in more writes downstream and it was all
inflicted on the same database clusters. Something had to give because this was
manifesting by customers seeing the replication lag with their eyes.
The only real paths out of this are to shard FoundationDB clusters or to just throw more hardware at the problem. Given that ram is at human kidney prices and that we have a small enough team that we're out of spare kidneys (for now, until we hire more), we gotta eat the complexity of setting up a second system to handle the message queue logic. We've had to cut our ram configurations in half. It's brutal out there.
So we ended up turning our message queue into a monstrous vermin.
Well, we didn't completely end the friendship with FoundationDB, we still use it across our workloads, but everything splits into three categories now:
- S3API or event triggered jobs: They go to Kafka.
- Jobs that need a scan to find: Think TTL expiry and lifecycle transitions. We still have FoundationDB scan for them, but then that scanner produces queue items in Kafka instead of FoundationDB.
- Replication: This stays in FoundationDB to avoid the dual-write problem. We have saved so much write pressure everywhere else that we bought the wiggle-room we need to handle the writes for replication.
Most of the savings come from that first category of changes, so let's dig into that.
Deletion isn't really deletion at our scale
One of the fun facts about distributed systems nonsense is that at a certain scale "deletion" stops actually removing data from the database. We end up putting a tombstone in the database and deferring the actual removal until later, mostly so that we can have soft-deletes be soft instead of hard. We clean things up in a few stages:
- We have a recurring job per bucket to clean things up.
- Each per-bucket task scans for tombstones and for every tombstone that's past the retention period, it schedules another task to actually clean that tombstone up.
- Each cleanup task then checks yet again and deletes the tombstone and its associated data blocks.
Let's do some napkin math. Every one of those tasks is a QuiCK job, and every QuiCK job costs an enqueue write, a claim write, at least one lease write, and a delete when it finishes. Here's what it takes to remove one tombstone, and what survives the move to Kafka:
| Write | QuiCK | Kafka |
|---|---|---|
| tombstone written over the key | ✓ | ✓ |
| cleanup task enqueued | ✓ | ✗ |
| cleanup task claimed | ✓ | ✗ |
| cleanup task leased, at least once | ✓ | ✗ |
| cleanup task deleted when it finishes | ✓ | ✗ |
| tombstone and its data blocks removed | ✓ | ✓ |
Four of those six writes are the queue talking to itself, and the per-bucket task that found the tombstone cost another four before it ever got there, plus range scans over the entire bucket. Then all the workers fighting for jobs are scanning over the work ranges, which competes with user requests against the storage servers. Delete a million objects and that's at least four million queue writes on a cluster that's probably already the bottleneck.
In our brave new Kafka world, stage 1 and 2 of that vanish. Servers write tombstones to FoundationDB, enqueue a message to a cleanup topic, and go back to doing whatever it is you want them to do. The cleanup topic is naturally in delete order, so consumers read from the earliest offset and process tombstones as their time comes.
Kafka's consumer offset ends up doing the job that QuiCK's vesting time did with no additional FoundationDB write pressure. The queue is the schedule. The only cost is that Kafka has its own transactions outside of FoundationDB. This is okay because the failure mode of the FoundationDB transaction working but the Kafka transaction failing is a tombstone sticking around longer than it would otherwise, not user data falling into the shadow realm. This tradeoff is acceptable.
So far nobody’s turned into a centipede
Overall, we’re pretty happy with how things have turned out. In an ideal world we’d be able to expand upon our FoundationDB-based message queue system. However, until we can put the time/energy into sharding our FoundationDB clusters, this works enough and should last us until we hit our next scaling threshold that requires us to rethink our design.
Oh, and as an added bonus we’re pretty sure that nobody on the team has turned into a centipede without warning. I’ll double-check with the team to be sure though!
Tigris is globally distributed, S3-compatible object storage. We handle the queues, the leases, and the clock skew so that you don’t have to.