Delay, Not Loss: FOQS, Meta's Trillion-Item Priority Queue

The Facebook ecosystem runs on thousands of distributed systems and microservices, and a large share of their work is better done asynchronously. Some of it is deferred to off-peak hours for better resource use, some scheduled for a future date, some just passed dependably between services. What all of those cases share is the need for a queue: a place to store work that must happen later, or elsewhere.

FOQS - Facebook Ordered Queueing Service - is that place for hundreds of use cases. Async, Facebook's general-purpose asynchronous compute platform, leans on FOQS to hold large backlogs and defer delay-tolerant work to off-peak hours. Video uploads are broken into components stored in FOQS for later encoding. Language translation splits computationally expensive jobs into parallel work items.

The scale sets the terms of the problem. FOQS processes close to a trillion items a day, and its backlogs have reached hundreds of billions of items. The post cites that number with pride, because a queue's backlog capacity is precisely its ability to absorb widespread downstream failure. When consumers break, the queue is the component that must not: producers keep enqueueing, items persist, and the failure surfaces as delay rather than loss.

But a priority queue makes a promise that gets harder as it scales on sharded storage. Items carry a user-specified 32-bit priority and a deliver_after timestamp, and dequeue must return the most important, oldest-ready work first. Yet the items live as rows spread across many MySQL shards, and each shard is owned by a single host and can see only its own rows. And MySQL itself pushes back against the queue's mechanics. The background operations that run FOQS - making deferred items deliverable, expiring leases, purging TTL'd items - are all timestamp scans, and MySQL retains old row versions in a history list whose length degrades read performance. The deeper the backlog, the longer the history, the slower the very queries that drain it. The queue's core engineering problem is keeping its performance independent of how full it is.

An item in FOQS is one row in a MySQL table, carrying a namespace, a topic, a priority (lower is more urgent), an immutable payload up to 10Kb, mutable metadata, a deliver_after timestamp, a lease duration, and a TTL. Its FOQS-assigned ID encodes the shard ID plus a 64-bit primary key, so every item's location is readable from its name. A topic is one logical priority queue, named by a string. Topics are cheap: name a new topic when you enqueue and it exists; drain its last item and it is gone. A namespace is the unit that separates tenants sharing the system. Each namespace is promised a guaranteed number of enqueues per minute and mapped to a tier - a pool of FOQS hosts and MySQL shards serving a group of namespaces. Shard Manager assigns each shard to exactly one host.

The enqueue path is buffered. A request lands on a host and immediately returns a promise, and a background worker for that shard writes the buffered items into MySQL rows, completing the promise once the insert lands. In front of each shard sits a circuit breaker, which the post cites by name. It marks a shard unhealthy on slow queries or elevated error rates over a rolling window and stops the worker from feeding it until it recovers, so FOQS does not keep piling new items onto an already-unhealthy shard. If an enqueue lands on an overloaded host, enqueue forwarding routes it to a host with capacity.

Dequeue is where the priority promise is kept. The API takes (topic, count) pairs and returns items ordered by priority, ties broken by older deliver_after. A host's items are spread across all the shards it owns, so to find the highest-priority work it must combine them. Each shard keeps a small in-memory index of its ready items, sorted by priority. The Prefetch Buffer merges those per-shard lists into one globally ordered stream in the background - a k-way merge, the standard way to combine several already-sorted lists. It then reads the winning rows, marks them 'delivered' so the same item can't go to two consumers, and stages them for the dequeue API. The buffer refills in proportion to how fast each topic is drained, so busier topics get more items staged ahead.

ONE PRIORITY ORDER, MANY SHARDS

Every delivery is handed out on a lease, which is just a countdown. When an item is dequeued, its clock starts. If the consumer confirms success (an ack) the row is deleted; if it reports failure (a nack) the item is rescheduled for later. But if the consumer does neither before the clock runs out - because it crashed, say - FOQS follows the topic's policy. At-least-once redelivers the item (no loss, but the consumer might see it twice); at-most-once deletes it (no duplicate, but that item is gone). A nack can carry a client-chosen delay, letting a failing consumer wait longer between attempts (exponential backoff), and can update the item's metadata to stash partial work. A lost ack is harmless too: if it never reaches FOQS, the lease expires and redelivery takes over.

FOQS is pull-based: consumers ask for work rather than having it pushed at them. The post defends that with a survey of its workloads - delays spanning milliseconds to days, consumption from tens of items a minute to more than ten million, per-topic and per-item priorities, and region affinity for some items. Push would make the queue responsible for not overwhelming every consumer; pull keeps the queue simple and lets each consumer set its own pace. The cost is that consumers must discover which topics have work, which FOQS answers with the getActiveTopics API and a routing layer.

Two mechanisms keep the queue fast and durable at depth. The first is checkpointing. FOQS's background jobs find work by scanning for rows whose timestamp is at or before now, but that scan forces MySQL to walk a list of old row versions as long as everything pending. A checkpoint is a saved marker of the last timestamp already processed; adding it as a lower bound means each scan covers only the new work since last time, so read performance stays flat no matter how deep the backlog. The second is disaster readiness: every MySQL shard is replicated asynchronously to two other regions, with the binlog (MySQL's change log) copied synchronously to a building nearby. Draining a datacenter then costs a few milliseconds of read-only while a replica catches up, is promoted, and the shard moves to a host in the new region. Namespace rate limits are enforced globally, not per-region - the post is direct that regional guarantees are impossible.