oxen-queue
v0.1.10
Published
A resilient worker queue backed by MySQL.
Downloads
393
Keywords
Readme
Oxen Queue
A no-frills, resilient worker queue backed by MySQL.
Features:
- Job persistence
- Job priority
- Job deduplication
- Concurrency
- Delayed jobs
- Multi-process/server operation
Motivation
Oxen is designed to help you chew through a very high number of jobs by leveraging significant concurrency. It is resilient to misbehaving jobs, dropped database connections, and other ills.
There are already several great job queue libraries out there, but in the context of our use-cases, they either struggled with a high number of jobs, handled unexpected disconnections poorly, or had issues with race conditions.
You'll be happy with Oxen if you:
- Have many, many jobs (millions per day isn't unreasonable)
- You're more interested in throughput than latency when it comes to job completion
- You want to be able to run arbitrary queries on the queue using SQL
- You're already running MySQL, and you don't want to add a another database to your stack (eg. Kafka)
Oxen isn't for you if:
- You need retry mechanisms for failed jobs
- Your jobs are user-facing and need to start in sub-second latencies
- You need a UI, and you don't want to hack something together yourself
- Using MySQL for a queue makes you feel icky
Installation
Infrastructure Requirements:
- Node 7 or higher
- MySQL
NPM
To install via npm, run:
npm install oxen-queue
Usage
Initialisation
Here's how you initialise the queue.
const Oxen = require('oxen-queue')
const ox = new Oxen({
mysql_config: {
user: 'mysql_user',
password: 'mysql_password',
// anything else you need to pass to the mysql lib
},
db_table: 'oxen_queue', // (optional) name the table that oxen will use in your database.
job_type: 'avatar_renders', // give this queue a job type. Other instances of oxen with the same job type will be the same queue.
})
/* If this is your first time running oxen, run this line to automatically create the database table. You should only need to run this once. */
await ox.createTable()
All constructor options that can be used when calling new Oxen({...})
:
| option | required? | default | type | description |
| -------------------- | --------- | ------------ | ------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| mysql_config | required | N/A | Connection Object | This object will be used to connect to your mysql instance. Use whatever you're already using to connect to mysql. At the minimum, you'll probably need {user: 'mysql_user', password: 'mysql_password'}
|
| db_table | optional | oxen_queue
| String | The table that Oxen will use to store its jobs. If you haven't specified a database name in your mysql_config
, you'll need to add your database as a prefix, such as my_database.oxen_queue
|
| job_type | required | N/A | String | The name of your queue, such as newsletter_emails
or user_sync
. Queues in other Node.js processes with the same name will share the same state. |
| extra_fields | optional | []
| Array | This array of strings allows you to add arbitary parts of your job body directly to your mysql table. Oxen will automatically pluck them out of your job body and insert them. It's up to you to alter your table to fit those extra fields. |
| fastest_polling_rate | optional | 100
| Int | The shortest delay between two polls of your table (ms) |
| slowest_polling_rate | optional | 10000
| Int | The longest delay between two polls of your table (ms) |
| polling_backoff_rate | optional | 1.1
| Int | The rate at which Oxen will slow polling if it finds no more jobs. For example, a rate of 1.2
will cause the next poll to be done 20% later than the last one. |
Adding Jobs
Jobs are added using addJob()
or addJobs()
const Oxen = require('oxen-queue')
const ox = new Oxen({ /* Initialisation args here */ }}
// adding a job with a string body
ox.addJob({
body : 'job_body_here'
})
// adding a job with an object body
ox.addJob({
body : { oh : 'hello', arr : [1, 2]}
})
// shorthand for adding a job with no additional parameters
ox.addJob('job_body_here')
// adding many jobs at once (batched insert)
ox.addJobs([
{ body : 'we' },
{ body : 'all' }
{ body : 'live' }
{ body : 'in' }
{ body : 'a' }
{ body : 'yellow' }
{ body : 'submarine' }
])
All addJob
options that can be used when calling addJob({...}
:
| option | required? | default | type | description |
| ---------- | --------- | ------------ | ---------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| body | required | N/A | Any | The job body. Will be JSON.stringify
'ed before saving to mysql. |
| unique_key | optional | null
| String/Int | Used for job deduplication. If you try to add two jobs with the same unique_key
, Oxen will discard the second one. This constraint is removed once the job finishes. |
| priority | optional | Date.now()
| Int | Defines the order that jobs will start processing. Smaller numbers will run first. Defaults to the current timestamp in milliseconds, so by default jobs will be popped fifo
. |
| start_time | optional | new Date()
| Date | Defines the time when Oxen will start trying to process this job. Accepts anything that new Date( ... )
does, such as ISO formatted strings, Date objects, and moment
objects. |
Consuming Jobs
Jobs are consumed using process()
.
const Oxen = require('oxen-queue')
const ox = new Oxen({ /* Initialisation args here */ }}
// start processing
ox.process({
work_fn : async function (job_body) {
// Do something with your job here
console.log(job_body)
return bigBadBackendThing(job_body.foo)
// The job will be considered finished when the promise resolves,
// or failed if the promise rejects.
}
concurrency : 25,
})
Your work_fn
will be called once per job that you added with addJob()
. It depends on promise resolution to know when the job is done, so make sure you return a promise!
Oxen will save the return of work_fn
in the result
field of the table.
If your jobs return large results, we recommend saving your actual result somewhere else in your infrastructure, and to return a small debugging marker such as "ok" or even null
or undefined
. This will keep the Oxen table from growing unnecessarily large.
If for any reason you want to stop processing jobs (for example, in the event of a graceful shutdown), call ox.stopProcessing()
All options that can be used with process()
:
| option | required? | default | type | description |
| ------------------ | --------- | ------- | -------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| work_fn | required | N/A | Async Function | Your work function. It only takes one argument (here job_body
), which is the body defined in addJob()
. It must return a Promise
. |
| concurrency | optional | 3 | Int | The number of jobs that Oxen will run at the same time. Higher numbers here allow Oxen to batch job fetches, increasing throughput. |
| timeout | optional | 60 | Int (seconds) | Jobs that don't return before the timeout elapses will be marked as failed. |
| recover_stuck_jobs | optional | true | Bool | If the process running Oxen is killed while jobs are still processing, jobs can get "stuck" in a processing state where Oxen no longer tries to run them. If recover_stuck_jobs
is true
, Oxen will check for stuck jobs every minute and put them back in a queued state. If it isn't safe to run a job twice, set this to false
. |
Performance
A few notes about Oxen's Performance.
- important Jobs are never removed from your database. It's up to you to clean them up when you no longer need their results or failure stacktraces. If you don't do this, your Oxen table may become very large! Even when very large (100GB+) it will still perform fine, but it becomes difficult to manually query anything that hasn't been carefully indexed.
- Assuming instantaneously-finishing jobs, the max throughput of Oxen depends on your
concurrency
andfastest_polling_rate
. Since Oxen batches job fetches with a size ofconcurrency
, an instance polling 10 times per second with aconcurrency
of 3 will at the maximum run 30 jobs per second. That said, if your jobs are so quick that you're limited by Oxen itself, Oxen may not be right for you. - Oxen will never query for another set of jobs if the previous query still hasn't returned. Nor will it try to query any more jobs if the there aren't any available
concurrency
slots. This means that you can set a very aggressivefastest_polling_rate
without hobbling your database -- afastest_polling_rate
of 2ms will never actually poll every 2ms, since mysql just doesn't query that fast! - Thanks to the
polling_backoff_rate
, queues without any jobs will quickly go back to theirslowest_polling_rate
. At atslowest_polling_rate
of10000
, Oxen will only query your database every 10 seconds. This means that after a period of inactivity, Oxen may take up to 10 seconds to start any new jobs that are added.
Internals
A big part of Oxen's appeal is that you can query it for your own uses. At the minimum, you'll probably want to query jobs for their results or failure messages.
In more advanced cases, you may want to add custom fields so that Oxen can be more tightly integrated into the rest of your application.
Data Storage
For your reference, here's the minimum schema of the table that Oxen uses:
CREATE TABLE IF NOT EXISTS `oxen_queue` (
`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
`batch_id` bigint(20) unsigned DEFAULT NULL,
`job_type` varchar(200) NOT NULL,
`created_ts` datetime DEFAULT CURRENT_TIMESTAMP,
`started_ts` datetime DEFAULT NULL,
`body` varchar(1000) DEFAULT NULL,
`status` varchar(100) NOT NULL DEFAULT 'waiting',
`result` mediumtext,
`recovered` tinyint(1) NOT NULL DEFAULT '0',
`running_time` smallint(5) unsigned DEFAULT NULL,
`unique_key` int(11) unsigned DEFAULT NULL,
`priority` bigint(20) DEFAULT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `unique_key` (`unique_key`),
KEY `created_ts` (`created_ts`),
KEY `status` (`status`),
KEY `locking_update` (`job_type`,`batch_id`,`status`,`priority`),
KEY `next_jobs_select` (`batch_id`,`priority`),
KEY `started_ts` (`started_ts`,`job_type`,`status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
Here's what the fields mean:
id
the job ID.batch_id
used to avoid race conditions when running Oxen on independent Node.js processes.job_type
a queue's indentifiercreated_ts
when the job was created and marked aswaiting
started_ts
when the job was marked asprocessing
and passed over to yourwork_fn()
body
the job bodystatus
determines the lifecycle of a job. Can bewaiting
,processing
,success
,error
,stuck
result
if the status issuccess
, it will contain the return value of yourwork_fn
in JSON. If the status iserror
, it will contain the error message and stacktracerecovered
default0
, will be set to1
if the job was recovered from astuck
status.running_time
the number of seconds during which the job wasprocessing
. Not actually read by Oxen, but useful for sanity checking.unique_key
used for job deduplication. Depends on a mysql unique index.priority
used for choosing which jobs to run first. Within ajob_type
, lower numbers will be processed first.
Extra Fields
If you add an extra column to your oxen_queue
table, Oxen will automatically populate that field for you based on what you pass into job_body
.
Here's an example. Imagine that you have a queue dedicated to updating your payment providers with your user metadata:
/*
This example assumes that you've added the fields user_id and payment_method to the oxen_queue table.
*/
const Oxen = require('oxen-queue')
// initialize a queue with an extra_fields array
const ox = new Oxen({
mysql_config: { ... },
job_type: 'sync_user_subscription_statuses',
extra_fields : ['user_id', 'payment_method']
})
// add a job with those extra_fields as keys in your job_body
ox.addJob({
body: {
user_id: 123,
payment_method: 'paypal',
some_other_thing : { whatever : 'value'}
}
})
// done! Your database table will now have the user_id and payment_method fields.
Note that user_id
and payment_method
will still also be available in job_body
.
Why add them as their own fields? They're already in the job_body...
Because you can index them! In our previous example, if you add an indexes to user_id
and payment_method
, you'll be able to query your table very effectively:
# Show all failing jobs for user_id 123
SELECT created_ts, started_ts, running_time, body, result
FROM oxen_queue
WHERE user_id = 123 AND STATUS = 'error';
# Show average running time per payment_method for jobs started in the last 6 hours.
SELECT payment_method, AVG(running_time)
FROM oxen_queue
WHERE started_ts > (NOW() - INTERVAL 6 HOUR)
GROUP BY payment_method;
This is where using an SQL-backed queue can really help debug tricky errors.
Authors
Oxen is written and maintained by the dev team at Opteo. Made with love in London.