FOQS: Scaling a distributed priority queue

Why Facebook Built a Distributed Priority Queue

Facebook’s infrastructure is made up of thousands of distributed systems and microservices, many of which benefit from running workloads asynchronously, especially during traffic peaks. Asynchronous execution allows for better resource usage, improved reliability, and the ability to defer computation to a later time. These workloads all share a common need: a queue to hold work that should run out-of-band or be passed between services.

Facebook Ordered Queueing Service (FOQS) is the internal solution for this. It is a fully managed, horizontally scalable, multitenant, persistent distributed priority queue built on sharded MySQL. FOQS lets developers at Facebook decouple their microservices and scale them independently.

The Core FOQS Workflow

A typical FOQS deployment has two actors:

• Producer: enqueues work items, optionally assigning a priority (lower number means higher priority) or a delay (to defer processing).
• Consumer: calls getActiveTopics to find topics with items, then dequeues items for processing. On success it acks them; on failure it nacks them, which triggers redelivery (with an optional delay).

FOQS handles hundreds of workloads across Facebook, including:

• Async, Facebook’s general-purpose asynchronous compute platform, which defers delay-tolerant workloads to off-peak hours.
• Video encoding, where uploaded videos are split into components, stored in FOQS, and processed independently.
• Language translation, where translation jobs are parallelized into many items and run by concurrent workers.

Core Data Model

FOQS organizes data into three levels: items live in topics, which live in namespaces. The service exposes a Thrift API with five operations: enqueue, dequeue, ack, nack, and getActiveTopics. Shard Manager assigns each MySQL shard to exactly one FOQS host.

Items

An item is a message in a priority queue. Each item is stored as a single row in a MySQL table and contains:

• Namespace: the multitenancy boundary.
• Topic: the logical queue the item belongs to.
• Priority: a user-specified 32-bit integer; lower values are delivered first.
• Payload: an immutable binary blob, up to 10 Kb.
• Metadata: a mutable binary blob, typically a few hundred bytes.
• Dequeue delay (deliver_after): earliest time the item can be dequeued.
• Lease duration: how long a consumer has to ack or nack after dequeue; otherwise FOQS redelivers based on the retry policy.
• Unique ID: a FOQS-assigned identifier used by the API.
• TTL: how long the item may live; expired items are deleted.

Topics and Namespaces

A topic is a logical priority queue referenced by a user-defined string. Topics are cheap and dynamic — creating one is just enqueuing an item to it. When a topic drains, it effectively disappears. The getActiveTopics API lets developers discover which topics currently have items.

A namespace maps to a specific use case and is the unit of multitenancy. Each namespace has guaranteed capacity, measured in enqueues per minute. Namespaces sharing a tier (a group of FOQS hosts and MySQL shards) don’t affect each other’s capacity, and each namespace belongs to exactly one tier.

Enqueue Path

Enqueue is the entry point for items. The request is buffered on the FOQS host, then a per-shard worker inserts the item as a row into MySQL. On success, the client receives a unique ID comprising the shard ID and a 64-bit primary key, which uniquely identifies the item across FOQS.

FOQS protects unhealthy shards with the circuit breaker pattern. Health is judged by slow queries (average beyond a threshold over a rolling window) or error rate (similarly averaged). If a shard is marked unhealthy, its worker stops accepting new work until the shard recovers.

Dequeue Path

The dequeue API accepts a collection of (topic, count) pairs and returns up to count items per topic. Items are delivered ordered by priority first, then by deliver_after (older first) for ties. Dequeue starts the item’s lease; if the consumer doesn’t ack or nack in time, the item becomes available for redelivery. FOQS supports both at-least-once (redeliver on lease expiry) and at-most-once (delete on lease expiry) semantics.

Because priorities matter, each host runs a reduce across its shards to surface the highest-priority items. FOQS optimizes this with a Prefetch Buffer: a background process that finds the top items across all shards and stages them for dequeue. Each shard holds an in-memory index of ready-to-deliver primary keys, sorted by priority, updated on operations like enqueue. The Prefetch Buffer performs a k-way merge to identify the best keys and issues select queries to fetch rows, marking them “delivered” to prevent double delivery.

The Prefetch Buffer sizes itself from observed client demand — topics being dequeued faster get replenished faster — so dequeue simply reads items straight from the buffer.

Ack and Nack

Ack confirms successful processing; the item is deleted. Nack requests redelivery; clients may add a delay to implement exponential backoff and can update metadata with partial results. Nacks update the existing row with a new deliver_after time rather than deleting it.

Routing is deterministic: each item ID encodes its shard, and a client uses the Shard Manager mapping to send the ack or nack to the host owning that shard. The operation goes into a shard-specific in-memory buffer with a worker that applies it to MySQL. If the operation is lost due to a host crash or MySQL unavailability, the item’s lease soon expires and it will be redelivered.

Why Pull Instead of Push

FOQS uses a pull-based API (consumers call dequeue) rather than pushing items out. The choice reflects the variation in FOQS workloads:

• Delay tolerance: items range from millisecond-critical to day-tolerant.
• Consumption rates: from tens of items per minute to over ten million per minute, with rates that vary independently of production rates based on downstream capacity.
• Priority structure: both at the topic level and per-item within a topic.
• Processing locality: some items must be handled in particular regions for data affinity.

A pull model with a routing layer handles these constraints better than push; it gives consumers full control over where and when they process items, while FOQS can still point them at active topics and available work.

Production Optimizations

FOQS has grown exponentially and currently processes close to one trillion items per day, with backlogs reaching hundreds of billions of items during widespread downstream failures. Handling this scale has required several targeted optimizations.

Bounding Background Scans

Background threads handle periodic work such as making deferred items ready, expiring leases, and purging expired items. These operations rely on timestamp columns — for instance, updating all ready-to-deliver items requires a query that selects every row where timestamp_column <= UNIX_TIMESTAMP() for update.

The issue is that MySQL locks updates to all rows matching roughly that condition, not just the ones returned. It retains old row versions in a linked structure called the history list; the longer that list grows, the slower read queries become.

With checkpointing, FOQS tracks a lower bound — the last known processed timestamp — and uses it to bound the query on both sides:

WHERE <checkpoint> <= timestamp_column AND timestamp_column <= UNIX_TIMESTAMP()

Restricting the range reduces the number of historical rows, improving both read and update performance.

Disaster Readiness and Routing

Facebook infrastructure must survive the loss of entire data centers. Each FOQS MySQL shard is replicated asynchronously to two additional regions, while the MySQL binlog is synchronously persisted to another building within the same region.

When a data center must be drained or MySQL is under maintenance, the primary is temporarily put into read-only mode until replicas catch up — usually a matter of milliseconds — then a replica is promoted. With the primary now in another region, the shard is reassigned to a FOQS host in that region to minimize expensive cross-region traffic.

Promotion events can create significant capacity imbalances across regions since FOQS cannot assume where capacity will be available. Routing has been improved so that enqueues go to hosts with capacity, and dequeues target hosts holding the highest-priority items. Two additional optimizations support disaster scenarios:

1. Enqueue forwarding: an enqueue request landing on an overloaded host is forwarded to one with available capacity.
2. Global rate limiting: since namespaces are the unit of multitenancy, each namespace gets a rate limit measured in enqueues per minute, enforced globally across regions. Guaranteeing per-region limits is not feasible, but traffic patterns help colocate capacity with demand and minimize cross-region traffic.

Current Priorities

FOQS supports hundreds of services across Facebook's stack, and reliable operation remains the top priority. Near-term work targets:

• Handling multi-domain failures (region, data center, rack): when a region goes down, FOQS must ensure no data loss and make data available from another region as quickly as possible.
• Improving load balancing for enqueue traffic and item discoverability during dequeues. New data centers spread data further worldwide, making discovery increasingly important for diverse traffic patterns.
• Expanding features for developer needs around workflows, timers, and strict ordering.

The FOQS team thanks all contributing engineers: Brian Lee, Dillon George, Hrishikesh Gadre, Jasmit Kaur Saluja, Jeffrey Warren, Manukranth Kolloju, Niharika Devanathan, Pavani Panakanti, Shan Phylim, and Yingji Zhang.