Ratus

Ratus is a RESTful asynchronous task queue server. It translated concepts of distributed task queues into a set of resources that conform to REST principles and provides a consistent HTTP API for various backends.

Programming language: Go

License: Apache License 2.0

Tags: Messaging    

README

Ratus

Ratus is a RESTful asynchronous task queue server. It translated concepts of distributed task queues into a set of resources that conform to REST principles and provides an easy-to-use HTTP API.

The key features of Ratus are:

• Guaranteed at-least-once execution of tasks.
• Automatic recovery of timed out tasks.
• Simple language agnostic RESTful API.
• Time-based task scheduling.
• Naturally load balanced across consumers.
• Support dynamic topology changes.
• Scaling through replication and partitioning.
• Pluggable storage engine architecture.
• Prometheus integration for observability.

Quick Start

Installation

Ratus offers a variety of installation options:

• Docker images are available on Docker Hub and GitHub Packages.
• Kubernetes and Docker Compose examples can be found in the deployments directory.
• Pre-built binaries for all major platforms are available on the GitHub releases page.
• Build from source with go install github.com/hyperonym/ratus/cmd/ratus@latest.

Depending on the storage engine you choose, you may also need to deploy the corresponding database or broker. Using the default mongodb engine as an example, assuming the database is already running locally, then start Ratus with:

$ ratus --port 8000 --mongodb-uri mongodb://127.0.0.1:27017

Basic Usage

Concepts introduced by Ratus will be bolded below, see Concepts (a.k.a cheat sheet) to learn more.

cURL

A producer creates a new task and pushes it to the example topic:

$ curl -X POST -d '{"payload": "hello world"}' "http://127.0.0.1:8000/v1/topics/example/tasks/1"

Example response

{
    "created": 1,
    "updated": 0
}

A consumer can then make a promise to claim and execute the next task in the example topic:

$ curl -X POST "http://127.0.0.1:8000/v1/topics/example/promises?timeout=30s"

Example response

{
    "_id": "1",
    "topic": "example",
    "state": 1,
    "nonce": "e4SN6Si1nOnE53ou",
    "produced": "2022-07-29T20:00:00.0Z",
    "scheduled": "2022-07-29T20:00:00.0Z",
    "consumed": "2022-07-29T20:00:10.0Z",
    "deadline": "2022-07-29T20:00:40.0Z",
    "payload": "hello world"
}

After executing the task, remember to acknowledge Ratus that the task is completed using a commit:

$ curl -X PATCH "http://127.0.0.1:8000/v1/topics/example/tasks/1"

Example response

{
    "_id": "1",
    "topic": "example",
    "state": 2,
    "nonce": "",
    "produced": "2022-07-29T20:00:00.0Z",
    "scheduled": "2022-07-29T20:00:00.0Z",
    "consumed": "2022-07-29T20:00:10.0Z",
    "deadline": "2022-07-29T20:00:40.0Z",
    "payload": "hello world"
}

If a commit is not received before the promised deadline, the state of the task is will be set back to pending, which in turn allows consumers to try to execute it again.

Go Client

Ratus comes with a Go client library that not only encapsulates all API calls, but also provides idiomatic poll-execute-commit workflows. The examples directory contains ready-to-run examples for using the library:

• The hello world example demonstrated the basic usage of the client library.
• The crawl frontier example implemented a simple URL frontier for distributed web crawlers. It utilized advanced features like concurrent subscribers and time-based task scheduling.

Concepts

Data Model

• Task references an idempotent unit of work that should be executed asynchronously.
• Topic refers to an ordered subset of tasks with the same topic name property.
• Promise represents a claim on the ownership of an active task.
• Commit contains a set of updates to be applied to a task.

Workflow

• Producer client pushes tasks with their desired date-of-execution (scheduled times) to a topic.
• Consumer client makes a promise to execute a task polled from a topic and acknowledges with a commit upon completion.

Topology

• Both producer and consumer clients can have multiple instances running simultaneously.
• Consumer instances can be added dynamically to increase throughput, and tasks will be naturally load balanced among consumers.
• Consumer instances can be removed (or crash) at any time without risking to lose the task being executing: a task that has not received a commit after the promised deadline will be picked up and executed again by other consumers.

Task States

• pending (0): The task is ready to be executed or is waiting to be executed in the future.
• active (1): The task is being processed by a consumer. Active tasks that have timed out will be automatically reset to the pending state. Consumer code should handle failure and set the state to pending to retry later if necessary.
• completed (2): The task has completed its execution. If the storage engine implementation supports TTL, completed tasks will be automatically deleted after the retention period has expired.
• archived (3): The task is stored as an archive. Archived tasks will never be deleted due to expiration.

Behavior

• Task IDs across all topics share the same namespace (ADR). Topics are simply subsets generated based on the topic properties of the tasks, so topics do not need to be created explicitly.
• Ratus is a task scheduler when consumers can keep up with the task generation speed, or a priority queue when consumers cannot keep up with the task generation speed.
• Tasks will not be executed until the scheduled time arrives. After the scheduled time, excessive tasks will be executed in the order of the scheduled time.

Engines

Ratus provides a consistent API for various storage engine implementations, allowing users to choose a specific engine based on their needs without having to modify client-side code.

Name	Persistent	Replication	Partitioning	TTL
mongodb	●	●	●	●

MongoDB

Ratus works best with MongoDB version ~4.4. MongoDB 5.0+ is also supported but requires additional considerations, see Implementation Details to learn more.

💭 TL;DR set MONGODB_DISABLE_ATOMIC_POLL=true when using Ratus with MongoDB 5.0+.

Replication

When using the MongoDB storage engine, the Ratus instance itself is stateless. For high availability, start multiple instances of Ratus and connect them to the same MongoDB replica set.

All Ratus instances should run behind load balancers configured with health checks. Producer and consumer clients should connect to the load balancer, not directly to the instances.

Partitioning

Horizontal scaling could be achieved through sharding the task collection. However, with the help of the TTL mechanism, partitioning is not necessary in most cases. The best performance and the strongest atomicity can only be obtained without sharding.

If the amount of data exceeds the capacity of a single node or replica set, choose from the following sharding options:

• If there is a large number of topics, use a hashed index on the topic field as the shard key, this will also enable the best polling performance on a sharded cluster.
• If there is a huge amount of tasks in a few topics, use a hashed index on the _id field as the shard key, this will also result in a more balanced data distribution.

Implementation Details

• When using the MongoDB storage engine, tasks across all topics are stored in the same collection.
• Task is the only concrete data model in the MongoDB storage engine, while topics and promises are just conceptual entities for enforcing the RESTful design principles.
• Since the resolution of the scheduled time in MongoDB is in millisecond level and is affected by the instance's own clock, the order in which consumers receive tasks is not strictly guaranteed.
• TTL cannot be disabled for completed tasks, in order to preserve a task forever, set it to the archived state.
• It is not recommended to upsert tasks on sharded collections using the topic field as the shard key. Due to MongoDB's own limitations, atomic operations cannot be used in this case, and only a fallback scheme equivalent to delete before insert can be used, so atomicity and performance cannot be guaranteed. This problem can be circumvented by using simple inserts in conjunction with fine-tuned TTL settings.
• By default, polling is implemented through findAndModify. In the event of a conflict, MongoDB's native optimistic concurrency control (OCC) will transparently retry the operation. But in MongoDB 5.0 and above, the retry will report a WriteConflict error in the database server's log (although the operation is still successful from the client's perspective). You can choose to ignore this error, or circumvent the problem by setting MONGODB_DISABLE_ATOMIC_POLL=true when using MongoDB 5.0+. This option will make Ratus to not use findAndModify for polling and instead rely on the application-level OCC layer to ensure atomicity.

Index Models

Key Patterns	Partial Filter Expression	TTL
{"topic": "hashed"}	-	-
{"topic": 1, "scheduled": 1}	{"state": 0}	-
{"deadline": 1}	{"state": 1}	-
{"topic": 1}	{"state": 1}	-
{"consumed": 1}	{"state": 2}	MONGODB_RETENTION_PERIOD

Observability

Metrics and Labels

Ratus exposes the following Prometheus metrics on the /metrics endpoint:

Name	Type	Labels
ratus_request_duration_seconds	histogram	topic, method, endpoint, status_code
ratus_chore_duration_seconds	histogram	-
ratus_task_schedule_delay_seconds	gauge	topic, producer, consumer
ratus_task_execution_duration_seconds	gauge	topic, producer, consumer
ratus_task_produced_count_total	counter	topic, producer
ratus_task_consumed_count_total	counter	topic, producer, consumer
ratus_task_committed_count_total	counter	topic, producer, consumer

Liveness and Readiness

Ratus supports liveness and readiness probes via HTTP GET requests:

• The /livez endpoint returns a status code of 200 if the instance is running.
• The /readyz endpoint returns a status code of 200 if the instance is ready to accept traffic.

Caveats

• 🚨 Topic names and task IDs must not contain plus signs ('+') due to gin-gonic/gin#2633.
• It is not recommended to use Ratus as the main storage of tasks. Instead, consider storing the complete task record in a database, and use a minimal descriptor as the payload for Ratus.
• The completed state only indicates that the task has been executed, it does not mean the task was successful.
• Ratus is a simple and reliable alternative to task queues like Celery. Consider to use RabbitMQ or Kafka if you need high-throughput message passing without task management.

Frequently Asked Questions

For more details, see Architectural Decision Records.

Why HTTP API?

Asynchronous task queues are typically used for long background tasks, so the overhead of the HTTP API is not significant compared to the time spent by the tasks themselves. On the other hand, the HTTP-based RESTful API can be easily accessed by all languages without using dedicated client libraries.

How to poll from multiple topics?

If the number of topics is limited and you don't care about the priority between them, you can choose to create multiple threads/goroutines to listen to them simultaneously. Alternatively, you can create a topic of topics to get the topic names in turn and then get the next task from the corresponding topic.

Contributing

This project is open-source. If you have any ideas or questions, please feel free to reach out by creating an issue!

Contributions are greatly appreciated, please refer to CONTRIBUTING.md for more information.

License

Ratus is available under the Mozilla Public License Version 2.0.

---

© 2022 Hyperonym