Design a Distributed Message Queue

1. Problem Statement & Scope Clarification

System Mission

Design a highly scalable, fault-tolerant, multi-tenant distributed message queue service (combining semantics of Amazon SQS and Apache Kafka/Amazon MSK) supporting asynchronous decoupling, durable at-least-once and strictly FIFO delivery semantics, configurable retention, dynamic visibility timeouts, and automated Dead-Letter Queue (DLQ) redrive capabilities.

Functional Requirements

1. Producer Ingestion (SendMessage / SendMessageBatch): Allow upstream microservices to produce single or batched messages with custom delay and deduplication keys.
2. Consumer Ingestion (ReceiveMessage): Allow downstream consumers to poll messages with configurable visibility timeouts and long polling (up to 20 seconds).
3. Acknowledgment & Deletion (DeleteMessage / DeleteMessageBatch): Explicitly acknowledge successful processing via cryptographic receipt handles.
4. FIFO & Partition Ordering: Guarantee strictly in-order, exactly-once processing per MessageGroupId while scaling horizontally across millions of groups.
5. Visibility Timeout Extension (ChangeMessageVisibility): Allow long-running worker tasks to heartbeat and extend execution leases to prevent double processing.
6. Dead-Letter Queue (DLQ) & Redrive: Automatically divert poison pills exceeding max retry limits (maxReceiveCount) to an isolated DLQ without halting partition progress.

Non-Functional Requirements (SLAs & SLOs)

• High Availability: $99.999\%$ uptime across multi-AZ AWS deployments; zero single point of failure.
• Durability: Zero acknowledged message loss ($\text{Durability} \ge 99.999999999\% / \text{11 Nines}$) achieved via synchronous multi-AZ replica quorum.
• Throughput & Scalability:
  • Standard Queues: Horizontally unlimited ($> 100,000\text{ msg/sec}$).
  • FIFO Queues: Up to $30,000\text{ msg/sec}$ with high-throughput batching enabled.
• End-to-End Latency: Publish-to-receive P99 Latency $< 15\text{ ms}$.

2. Capacity & Scale Estimation (Back-of-the-Envelope Math)

Ingestion & Egress Throughput

• Daily Ingestion Volume: $2.0 \times 10^9\text{ messages/day}$ (2 Billion messages/day).
• Average Message Size: $2\text{ KB}$ (Max payload: $256\text{ KB}$).
• Average Ingest QPS: $\text{Average Ingest QPS} = \frac{2 \times 10^9\text{ messages}}{86,400\text{ sec}} \approx \mathbf{23,148\text{ msg/sec}}$
• Peak Ingest QPS ($3.5\times$ peak-to-average ratio): $\text{Peak Ingest QPS} = 23,148 \times 3.5 \approx \mathbf{81,000\text{ msg/sec}}$
• Ingest Bandwidth: $\text{Peak Ingest Bandwidth} = 81,000\text{ msg/sec} \times 2\text{ KB} = 162\text{ MB/sec} \approx \mathbf{1.296\text{ Gbps}}$
• Fan-Out Egress ($2\times$ average consumer groups): $\text{Peak Egress Bandwidth} = 1.296\text{ Gbps} \times 2 = \mathbf{2.592\text{ Gbps}}$

Storage Footprint & Replication Capacity

• Raw Daily Storage Ingested: $2 \times 10^9\text{ msg/day} \times 2\text{ KB} = \mathbf{4\text{ TB/day}}$
• 7-Day Retention Storage: $\text{7-Day Raw Storage} = 4\text{ TB/day} \times 7\text{ days} = 28\text{ TB}$
• 3-Way Multi-AZ Storage Replication + $20\%$ Index Overhead: $\text{Total Cluster Storage} = 28\text{ TB} \times 3 \times 1.20 = \mathbf{100.8\text{ TB}}$

3. High-Level Architecture & AWS Component Mapping

Interactive Architecture Diagram
100%Fit100%
Synthesizing vector architecture diagram...

4. API Interface Design & Wire Protocol

Core HTTP/gRPC API Specifications

protobufCopy
syntax = "proto3";

package hispeeddesign.queue.v1;

service MessageQueueService {
  rpc SendMessage (SendMessageRequest) returns (SendMessageResponse);
  rpc SendMessageBatch (SendMessageBatchRequest) returns (SendMessageBatchResponse);
  rpc ReceiveMessage (ReceiveMessageRequest) returns (ReceiveMessageResponse);
  rpc DeleteMessage (DeleteMessageRequest) returns (DeleteMessageResponse);
  rpc ChangeMessageVisibility (ChangeMessageVisibilityRequest) returns (ChangeMessageVisibilityResponse);
}

message SendMessageRequest {
  string queue_name = 1;
  string message_body = 2;
  int32 delay_seconds = 3;
  string message_group_id = 4;           // Mandatory for FIFO queues
  string message_deduplication_id = 5;   // SHA-256 or custom deduplication token
  map<string, string> message_attributes = 6;
}

message SendMessageResponse {
  string message_id = 1;
  string sequence_number = 2;            // Strict monotonic 64-bit sequence in FIFO
  string md5_of_body = 3;
}

message ReceiveMessageRequest {
  string queue_name = 1;
  int32 max_number_of_messages = 2;      // 1 to 10
  int32 visibility_timeout_seconds = 3;   // Default: 30s, Max: 43,200s (12 hours)
  int32 wait_time_seconds = 4;           // Long polling timeout: 0 to 20s
}

message ReceiveMessageResponse {
  repeated QueueMessage messages = 1;
}

message QueueMessage {
  string message_id = 1;
  string receipt_handle = 2;             // Ephemeral cryptographic lease token
  string message_body = 3;
  int32 receive_count = 4;               // Incremented on each consume attempt
  int64 publish_timestamp_ms = 5;
  string message_group_id = 6;
}

message DeleteMessageRequest {
  string queue_name = 1;
  string receipt_handle = 2;
}

message DeleteMessageResponse {
  bool success = 1;
}

message ChangeMessageVisibilityRequest {
  string queue_name = 1;
  string receipt_handle = 2;
  int32 visibility_timeout_seconds = 3;
}

message ChangeMessageVisibilityResponse {
  bool success = 1;
  int64 new_visibility_deadline_ms = 2;
}

5. Storage Engine Internals & Zero-Copy Architecture

1. Append-Only Segment Log Structure

Every topic partition stores messages sequentially inside fixed-size segment files (1\text{ GB} each) on disk:

• .log (Segment Data): Sequential binary stream of messages [Length | CRC32 | MagicByte | Attributes | Timestamp | KeyLength | Key | ValLength | Value].
• .index (Sparse Offset Index): Maps logical message offset to physical byte offset in .log (1 entry every $4\text{ KB}$ of log data).
• .timeindex (Sparse Timestamp Index): Maps epoch millisecond timestamp to logical message offset for time-based seeks and TTL expiry.

textCopy
Disk Segment 00000000000000000000:
+------------------------------------------------------------------------------------+
| Offset 0 | Msg: "order_1" | Offset 1 | Msg: "order_2" | Offset 2 | Msg: "order_3"  |
+------------------------------------------------------------------------------------+

Sparse Index (.index):
+------------------------------------+
| Logical Offset 0   -> Byte Offset 0|
| Logical Offset 100 -> Byte Offset 4096 |
| Logical Offset 200 -> Byte Offset 8192 |
+------------------------------------+

2. Zero-Copy Kernel Data Path (sendfile)

To eliminate CPU and memory bus bottlenecks during high-throughput consumer egress, the broker relies on the Linux sendfile(2) system call:

Interactive Architecture Diagram
100%Fit100%
Synthesizing vector architecture diagram...