nTorque - web hook task queue

nTorque is a task queue service that uses web hooks. It is free, open source software released into the public domain that you can use from any programming language (that speaks HTTP) to queue up and reliably execute idempotent tasks. For example, in Python:

import os
import requests

params = {'url': 'http://example.com/myhooks/send_email'}
data = {'user_id': 1234}

endpoint = os.environ.get('NTORQUE_URL')
response = requests.post(endpoint, data=data, params=params)

Rationale

nTorque is designed to be a good solution when you need more reliability than fire-and-forget but you don't need an AMPQ / ESB sledgehammer to crack your "do this later" nut.

Because it uses web hooks, you can:

1. use it from (and to integrate) applications written in any language
2. use DNS / web server load balancing to distribute tasks
3. bootstrap your task execution environment the way you bootstrap a web application -- i.e.: once at startup, re-using your web app's configuration and middleware

Functionality

nTorque provides the following endpoints:

• POST / to enqueue a task
• GET /tasks/:id to view task status

And the following features:

• persistent task storage
• non-blocking, concurrent task execution
• HTTPS and redirect support
• configurable (linear or exponential) backoff

Implementation

nTorque is a Python application comprising of a web application and one or more worker processes. These use a PostgreSQL database to persist tasks and a Redis database as a notification channel.

+------+  |  +--------+    +--------+    +--------+  |
|POST /|     |Frontend|    |Web app |    |Postgres|
|------|  |  |--------|    |--------|    |--------|  |
|- url |+- ->|- auth  |+-->|- store |+-->|- tasks |
|- data|  |  |- rate  |    |- notify|    |        |  |
|      |     |  limits|    |        |    |        |
+------+  |  +--------+    +--------+    +--------+
                               +           ^    +    |
          |                    |           |   url
                             rpush        get  data  |
          |                    |           |    |
                               v           +    v    |
          |                 +--------+   +--------+     +---------+
                            |Redis   |   |Worker  |  |  |Web hook |
          |                 +--------+   |--------|     |---------|
                               |         |- POST  |+-|->|- perform|
          | nTorque            +-blpop-> |  data  |     |  task   |
                                         +--------+  |  +---------+

In the event of a response with status code:

• 200 or 201: the task is marked as successfully completed
• 202 - 499: the task is marked as failed and is not retried
• 500 (or network error): the task is retried

Hack here if you'd like a different strategy.

Algorithm

The real crux of nTorque is a trade-off between request timeout and retry delay. It's worth understanding this before deploying -- and how to simply mitigate it by a) specifying an appropriate default timeout and b) overriding this as necessary on a task by task basis.

Like RQ and Resque, nTorque uses Redis as a push messaging channel. A request comes in, a notification is rpushd onto a channel and blpopd off. This means that tasks are executed immediately, with a nice evented / push notification pattern.

Unlike RQ and Resque, nTorque doesn't trust Redis as a persistence layer. Instead, it relies on good-old-fashioned PostgreSQL: the first thing nTorque does when a new task arrives is write it to disk. It then notifies a consumer process using Redis BLPOP. The consumer then reads the data from disk and performs the task by making an HTTP request to its url.

In most cases, this request will succeed, the task will be marked as completed and no more needs to be done. However, this won't happen every time, e.g.: when there's a network error or the webhook server is temporarily down. Because there are edge case failure scenarios where the web hook response is unreliable, nTorque refuses to rely on it as the source of truth™ about a task's status. Instead, the single source of truth is the PostgreSQL database.

This is achieved by automatically setting a task to retry every time it's read ("acquired") from the database. Specifically, the query that reads the task data is performed within a transaction that also updates the task's due date and retry count. This means that in any failure scenario, nTorque can always just be restarted (potentially on a new server as long as it connects to the same database) and you can be sure that tasks will be performed at least once no matter where they were in the pipeline when whatever it was fell over.

Incidentally, tasks due to be retried are picked up by a background process that polls the database every NTORQUE_REQUEUE_INTERVAL seconds.

More importantly, and where this description has been heading, is the relation between the due date of the task as it lies, gloriously in repose, and the timeout of the web hook call. For there is one thing we don't want to do, and that is keep retrying tasks before they've had a chance to complete.

In order to prevent this behaviour, we impose a simple constraint:

The due date set when the task is transactionally read and incremented must be longer than the web hook timeout.

This means that, in the worst case (when a web hook request does timeout or fail to respond), you must wait for the full timeout duration before your task is retried. So whilst you may naturally want to set a relatively high timeout for long running tasks, you may want to keep it shorter for simper tasks like sending your new user's welcome or reset password email: so that they're retried faster.

The good news is that, in addition to the global NTORQUE_DEFAULT_TIMEOUT configuration variable, you can set an appropriate timeout for different tasks using the timeout query parameter.

Simple -- once you know how the system works.

Installation

Clone the repo, install the Python app using:

bash pip_install.sh

You need Redis and Postgres running. If necessary, create the database:

createdb -T template0 -E UTF8 ntorque

If you like, install Foreman, to run the multiple processes, using:

bundle install

Run the migrations:

foreman run alembic upgrade head

Bootstrap an app (if you'd like to authenticate access with an API key):

foreman run python alembic/scripts/create_application.py --name YOURAPP

You should then be able to:

foreman start

Alternatively, skip the Foreman stuff and run the commands listed in Processes manually / using a Docker / Chef / init.d wrapper. Or push to Heroku, run the migrations and it should just work.

Configuration

Algorithm / Behaviour:

• NTORQUE_BACKOFF: exponential (default) or linear
• NTORQUE_CLEANUP_AFTER_DAYS: how many days to leave tasks in the db for, defaults to 7
• NTORQUE_DEFAULT_TIMEOUT: how long, in seconds, to wait before treating a web hook request as having failed -- defaults to 60 see the algorithm section above for details
• NTORQUE_MIN_DUE_DELAY: minimum delay before retrying -- don't set any lower than 2
• NTORQUE_MAX_DUE_DELAY: maximum retry delay -- defaults to 7200 but you should make sure its longer than NTORQUE_DEFAULT_TIMEOUT
• NTORQUE_MAX_RETRIES: how many attempts before giving up on a task -- defaults to 36
• NTORQUE_REQUEUE_INTERVAL: how often, in seconds, to poll the database for tasks to requeue -- defaults to 5
• NTORQUE_TRANSIENT_REQUEST_ERRORS: 4xx errors which ntorque should retry -- defaults to '408,423,429,449'

Deployment:

• NTORQUE_AUTHENTICATE: whether to require authentication; defaults to True -- see authentication section in Usage below
• NTORQUE_ENABLE_HSTS: set this to True if you're using HSTS
• HSTS_PROTOCOL_HEADER: set this to, e.g.: X-Forwarded-Proto if you're running behind an https proxy frontend (see pyramid_hsts for more details)
• MODE: if set to development this will run Gunicorn in watch mode (so the app server restarts when a Python file changes) and will raise HTTP exceptions in the API views (rather than returning them). If set to production it will run Gunicorn behind a newrelic client. If this isn't quite what you want then either don't set it or set it to any other string (or hack the run.sh and / or gunicorn.py scripts)

Redis:

• NTORQUE_REDIS_CHANNEL: name of your Redis list used as a notification channel; defaults to ntorque
• REDIS_URL, etc.: see pyramid_redis for details on how to configure your Redis connection

Database:

• DATABASE_URL, defaults to postgresql:///ntorque
• SQLALCHEMY_MAX_OVERFLOW, SQLALCHEMY_POOL_CLASS, SQLALCHEMY_POOL_SIZE and SQLALCHEMY_POOL_RECYCLE -- see the SQLAlchemy docs on engine configuration and pyramid_basemodel for more information; if you don't provide these then SQLAlchemy will use sensible defaults, also note that if you're using pgbouncer you should set SQLALCHEMY_POOL_CLASS=sqlalchemy.pool.NullPool

Usage / API

Authentication

If you set NTORQUE_AUTHENTICATE to True then you need to create at least one application (e.g.: using the alembic/scripts/create_application.py script) and provide its api key in the NTORQUE_API_KEY header when enqueuing a task.

POST /

To enqueue a task, make a POST request to the root path of your nTorque installation.

Required:

• a url query parameter; this is the url to your web hook that you want nTorque to call to perform your task

Optional:

• a method query parameter; which http method to use when calling the webhook -- the default is POST, but you can alternatively specify DELETE, PUT or PATCH.
• a timeout query parameter; how long, in seconds, to wait before treating the web hook call as having timed out -- see the Algorithm section above for context

Data:

This aside, you can pass through any POST data, encoded as any content type you like. The data, content type and character encoding will be passed on in the POST (or DELETE, PUT or PATCH) request to your web hook.

Headers:

Aside from the content type, length and charset headers, derived from your request, you can specify headers to pass through to your web hook, by prefixing the header name with NTORQUE-PASSTHROUGH-. So, for example, to pass through a FOO: Bar header, you would provide NTORQUE-PASSTHROUGH-FOO: Bar in your request headers.

Response:

You should receive a 201 response with the url to the task in the Location header.

GET /task/:id

Returns a JSON data dict with status information about a task.

POST /task/:id/push

Pushes a task onto the redis notification channel to be consumed, aquired and performed. You should not normally need to use this. It's exposed as an optimisation for hybrid integrations.

Pro-Tips

nTorque is a system for reliably calling web hook task handlers: not for implementing them. You are responsible for implementing and exposing your own web hooks. In most languages and frameworks this is simple, e.g.: in Ruby using Sinatra:

post '/hooks/foo' do
    # your code here
end

Or in Python using Flask:

@app.route('/hooks/foo', methods=['POST'])
def foo():
    # your code here

Key things to bear in mind are:

Status Code

After successfully performing their task, your web hooks are expected to return an HTTP response with a 200 or 201 status code. If not, nTorque will keep retrying the task.

Avoid Timeouts

Your web server must be configured with a high enough timeout to allow tasks enough time to complete. If not, you may be responding with an error when tasks are actually being performed successfully.

For example, for a 30 minute timeout with Apache as a proxy:

Timeout 1800
ProxyTimeout 1800

Or with Nginx:

send_timeout 1800;
proxy_send_timeout 1800;

Secure Public Hooks

If your web hooks are exposed on a public IP, you are likely to want to secure them, e.g.: using HTTPS and an authentication credential like an API key.

It's also worth noting that you may need to turn off CSRF validation.

Support

Raise bugs / issues on GitHub.