Design a Distributed Priority Queue

Today, we'll develop a Distributed Priority Queue, focusing on its implementation using a sharded SQL database. Our discussion will include:

• API and message schema
• Adding tasks to the queue (Enqueue)
• Removing tasks from the queue (Dequeue)
• Deploying across specific regions (Regional Deployment)
• Expanding the deployment to multiple regions (Multi-regional)

API, Data Structure for Items, and Status of Messages

The message schema outlines the data format that will be entered into and retrieved from the queue.

Namespace: This represents the isolation boundary for different tenant environments.

Topic: Acts as a logical queue; a single namespace can have numerous topics.

Priority: 32-bit integer where a lesser value denotes higher message urgency.

Payload: An immutable binary data block, limited by size.

Metadata: Mutable key-value data intended for ancillary information

Deliver After: The scheduled time for a message to be made available for consumption

Lease Duration: The allocated time frame within which a consumer must acknowledge (ack) or reject (nack) a dequeued message.

Unique ID: A unique identifier.

TTL: Defines the duration a message remains in the queue before automatic deletion

Enqueue operation

Enqueue is an operation to push a message to a Priority Queue. If an enqueue succeeds, the item is persisted and can eventually be dequeued.

Submit to Enqueue Buffer

• The serialized item is sent to the Priority Queue system, where it first lands in an enqueue buffer. This buffer acts as a preliminary holding area to manage incoming requests and mitigate spikes in traffic.
• The enqueue buffer ensures rate limiting and initial validation, preventing system overload and filtering out malformed requests before they reach the core processing units (enqueue workers).

Process with Enqueue Workers

• Enqueue workers continuously poll the enqueue buffer for new items. Once an item is picked up by a worker, it undergoes further validation, such as checking for correct formatting and ensuring priority levels are within acceptable ranges.
• These workers are responsible for the logical handling of items, including determining the appropriate SQL shard based on the item's attributes.

Distribute to SQL Shards

• After processing, the item is assigned to a specific SQL shard. The distribution can be based on various factors like load balancing, shard health, and data partitioning strategies
• Each SQL shard corresponds to a segment of the overall data structure, designed to scale horizontally and handle high volumes of writes. The enqueue worker inserts the item into the designated shard

Acknowledge and Log

• Once the item has been successfully inserted into the appropriate SQL shard, the system generates a unique identifier for the item, combining the shard ID with a unique item identifier within that shard

Dequeue operation

The dequeue API accepts a collection of (topic, count) pairs. For each topic requested, host will return, at most, count items for that topic. The items are ordered by deliver_after and priority, so items with current deliver_after and lower priority will be delivered first. If multiple items are tied for lowest priority, lower deliver_after items will be delivered first. 

Dequeueing items typically involves the following steps:

1. Request Dequeue: A dequeue request is initiated by a client or a service. This request specifies the number of items to be dequeued and may include priority or other criteria to select the appropriate items from the queue.
2. Select Items: The Queue evaluates the priority queue to identify the highest-priority items available for dequeuing. This involves sorting or retrieving items based on their assigned priorities.
3. Lock Items: To prevent race conditions or duplicate processing, the system temporarily locks the selected items. This ensures that no other process can dequeue or modify these items while they are being processed for removal.
4. Remove Items and Update Queue: Once the items are locked and confirmed for dequeuing, they are removed from the priority queue. The Queue system updates indexes, counters, an other metadata in the corresponding shard
5. Return Items and Unlock: The dequeued items are packaged and sent back to the client. After the items have been successfully returned, locks are released.

Regional Deployment

During any event leading to the unavailability of a primary replica in Region X, SQL DB can elect secondary replicas from Region 2 as the new primary. After failover, the queue service in Region 1 must send queries to the new primary database in Region 2 where cross-region latencies increase up to hundreds of milliseconds.

In the event of complete network connectivity loss in region 1, the limitations of the regional architecture become more glaring: 

• Customers have to explicitly balance their traffic away from an impacted region. Regional installations force all of the disaster recovery and load-balancing complexity to clients.
• Pushing load-balancing complexity to customers often results in underutilization of system resources. Additionally, it requires buffer capacity in other regions to accommodate shifted traffic.
• Due to an inability for SQL primaries to be picked up outside of a given regional installation, items stored on replicas in the primary regions are susceptible to being “stuck” until connectivity is restored. Regional installations fail to offer access to data with high availability.

Multi Regional Deployment

To enhance latency, we introduce a routing service acting as an intermediary between clients and the queue service. This service conceals the complexities of physical routing from clients and facilitates efficient distribution of messages

Routing service optimizes data allocation according to regional preferences. It uses in-memory mapping to associate logical regional preferences with the nearest SQL shards. The API allows clients to indicate their preferred storage regions.

When processing a request, Routing Service references its in-memory data to identify suitable queue nodes based on the client's regional choice. Consequently, items are stored in the designated physical location or an alternative region overseeing the relevant SQL shards in case of failover. If there's no region preference, items default to the nearest region from where the request originated.