# What is BullMQ Proxy BullMQ Proxy is a small and thin service that allows interacting with [BullMQ](https://bullmq.io) queues from a standard HTTP Restful API. BullMQ is originally written in NodeJS and uses Redis™ (or any other Redis™ compatible database). The Proxy supports all the powerful features provided by BullMQ that folks love and allows those features to be utilized through HTTP requests. This can significantly simplify integration with systems that are not based on Node.js or when direct interaction with Redis™ is not feasible or preferred. By acting as a bridge, BullMQ Proxy translates HTTP requests into the appropriate BullMQ and Redis commands, enabling applications to enqueue jobs, manage queues, and subscribe to events with simple HTTP calls. This means you can perform tasks like adding jobs to queues, fetching job statuses, and managing job lifecycles without having to write any Node.js code or directly interact with Redis. This approach offers several benefits: * **Language Agnostic**: It opens up the possibility of using BullMQ with any programming language that can make HTTP requests, broadening the accessibility of BullMQ's features to a wider range of developers and applications. * **Simplified Architecture**: By encapsulating the BullMQ logic within a RESTful interface, it simplifies the architecture of applications that need to interact with job queues, especially when those applications are built using microservices. * **Ease of Integration**: Integrating with external systems, webhooks, or third-party services becomes easier as HTTP is a universally supported protocol, and there's no need to deal with the complexities of direct Redis connections or specific BullMQ APIs. * **Scalability and Security**: The proxy can be scaled independently of the applications using it and can also serve as a security layer, controlling access to the job queues and Redis data. However, it's important to note that adding a proxy layer could introduce additional latency and a potential point of failure in your system architecture. Proper deployment, monitoring, and scaling practices are essential to mitigate these risks and ensure that the advantages of using BullMQ Proxy are maximized. BullMQ Proxy is written with new Javascript runtime [BunJS](https://bun.sh/). This runtime provides a very fast HTTP server stack for minimal overhead compared to running native BullMQ libraries. {% hint style="info" %} Note that there are some advanced use cases in BullMQ that are only available if working directly with the NodeJS library. {% endhint %} # Getting started BullMQ Proxy is delivered as a docker image, you can find all the images in Githubs container registry: {% embed url="" %} BullMQ Proxy container registry {% endembed %} You can the proxy directly by starting a container based on the image, but the easiest way is to use the following docker-compose.yaml: ```yaml version: '3' services: proxy: image: ghcr.io/taskforcesh/bullmq-proxy:latest ports: - 8080:8080 environment: PORT: 8080 REDIS_HOST: redis REDIS_PORT: 6379 REDIS_PASSWORD: ${REDIS_PASSWORD} REDIS_TLS: ${REDIS_TLS} AUTH_TOKENS: ${AUTH_TOKENS} redis: image: 'redis:alpine' ``` You can also have a local .env file with some secrets so that you can start testing the proxy, for example: ``` AUTH_TOKENS=1234 ``` Will set a test token "1234" that you can use to authenticate when interacting with the proxy. There are several environment variables that can be used to configure different aspects of the proxy. You can find all of them [here](/http-api/configuration). You can start the proxy now with `docker-compose up:` ``` $ docker-compose up WARN[0000] The "REDIS_PASSWORD" variable is not set. Defaulting to a blank string. WARN[0000] The "REDIS_TLS" variable is not set. Defaulting to a blank string. [+] Running 2/2 ✔ Container bullmq-proxy-proxy-1 Created 0.0s ✔ Container bullmq-proxy-redis-1 Created 0.0s Attaching to proxy-1, redis-1 redis-1 | 1:C 28 Feb 2024 14:11:32.929 # WARNING Memory overcommit must be enabled! Without it, a background save or replication may fail under low memory condition. Being disabled, it can also cause failures without low memory condition, see https://github.com/jemalloc/jemalloc/issues/1328. To fix this issue add 'vm.overcommit_memory = 1' to /etc/sysctl.conf and then reboot or run the command 'sysctl vm.overcommit_memory=1' for this to take effect. redis-1 | 1:C 28 Feb 2024 14:11:32.934 * oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo redis-1 | 1:C 28 Feb 2024 14:11:32.934 * Redis version=7.2.1, bits=64, commit=00000000, modified=0, pid=1, just started redis-1 | 1:C 28 Feb 2024 14:11:32.934 # Warning: no config file specified, using the default config. In order to specify a config file use redis-server /path/to/redis.conf redis-1 | 1:M 28 Feb 2024 14:11:32.940 * monotonic clock: POSIX clock_gettime redis-1 | 1:M 28 Feb 2024 14:11:32.972 * Running mode=standalone, port=6379. redis-1 | 1:M 28 Feb 2024 14:11:32.982 * Server initialized redis-1 | 1:M 28 Feb 2024 14:11:32.993 * Loading RDB produced by version 7.2.1 redis-1 | 1:M 28 Feb 2024 14:11:32.995 * RDB age 228268 seconds redis-1 | 1:M 28 Feb 2024 14:11:32.995 * RDB memory usage when created 0.88 Mb redis-1 | 1:M 28 Feb 2024 14:11:32.995 * Done loading RDB, keys loaded: 0, keys expired: 0. redis-1 | 1:M 28 Feb 2024 14:11:32.995 * DB loaded from disk: 0.012 seconds redis-1 | 1:M 28 Feb 2024 14:11:32.996 * Ready to accept connections tcp proxy-1 | $ bun src/index.ts proxy-1 | proxy-1 | ; proxy-1 | ;: ;: proxy-1 | :; :;++;. ::. ;: proxy-1 | .:;. :X&X$&$$$;$&&&X+: .;: proxy-1 | ::$XxX$$&&&&&&&;+X$$&$++XX$XX++$$x+;:: :x.: proxy-1 | :::+XXXx+x+++++:XXXX$$+;X$X$++:XXXX$$&&&&&&&&X:;. proxy-1 | :;:::::;;:. +++$$X$X;xXXXX::..:...::;+++;::; proxy-1 | ;+::. ;.:XX+XxXX;X$$Xx;.:.:;X: ... proxy-1 | x:::. :.++:X$XxxX$$xx :; proxy-1 | +$Xx: .XX+.:++$$$x;;:; .+X&: proxy-1 | X$x;:::+;+;.+X$$$X;:::.::+X&$. proxy-1 | xX+;+::+;x++++X$$x:;:;.;.:+X+XX; proxy-1 | xx:+:;.;;.++++$$$x;;;.:.;:.xx;++$$x::;;;. proxy-1 | x+;;;:::;+;X$xXXXXX: :::.::+$x .:+$&&&&&&&; proxy-1 | +$;;;;:::. +x:+++++;X .:::.+:+x$+&&X$;XX;++$&&; proxy-1 | +$X+::;::::.;+:: ::X ::. :.;$&x+$$X&$$&&; +$x++ proxy-1 | ;&$$X::;:; ::::.::: .:: ::;X$&$X+$:+&Xx$$+ :.;X: proxy-1 | X$X$$+:.::: . ::...++XX$Xx;X:X&$++x::XX;+++ proxy-1 | .X$;x+;::.+:. : . .::;x+x:XXx:::X$$+++ :+X$;+x+ proxy-1 | +Xx:X; :::::: :. : :: +;+X+;.+++. x+$X+++.X$+;:;xX: proxy-1 | ;:;: ::.:. : :. . :::;+++:. ::. :;:+;;+:.X&$x;.;x; proxy-1 | :..:.::. : ... ...:;;::. :; ::.:.;+:..+XX;:::;: proxy-1 | :;.:::.:. .. .. .:::;. .;x$Xx ..::.. :;:.;+x:. proxy-1 | $&;. .:. :: :: .. :. ;x$x:$; .::. . ::+;:: proxy-1 | :&x++. .:. ;: ::: ::+$X:.x; .:: ... ..:. ::+ proxy-1 | $;+X;..: . . +: .. .+x;Xx:;.:. . . ..: .:: :+ proxy-1 | ;+:++; .:. :. ;. .;:.::X;.; .... . .:;; :x proxy-1 | :;:.. . ::: ..;.;: ;.. .: : . .:::XX. :x: proxy-1 | :::.:. .::. .:+::.::;. :: : ;. .:::: :+ proxy-1 | .;;:... ::::... :+Xx. ::.;;: . + ::.. ... +x; proxy-1 | +$X; :+$$X:. :::. :++ .:: ++X++ proxy-1 | ;+::.;. :+. ;++: ::. :+: :: .:: :x;.;;. proxy-1 | ;:... :::.:; .x:.: : .:.::;+;:+ proxy-1 | ;..: ::.. X;.. :.; ::;;;;; proxy-1 | :X;:. :x+:. +x:: ::; ;;:+: proxy-1 | .$$x:: .$$X:: +X:: :::; ;;;;. proxy-1 | .;;;::.. +;;;:: x+;:: :::::: .;:; proxy-1 | .X:Xx;.. X+.x+;: +;+;: .+:;+; ..: proxy-1 | .X;XX;::: X+:xX+:. x;;+:. X;:x;:: proxy-1 | .:;+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++:.. proxy-1 | ..::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::: proxy-1 | [1709129499052][INFO] BullMQ Proxy: Running BullMQ Proxy on port 8080 (c) 2024 Taskforce.sh Inc. v0.1.0 ``` The proxy will now be ready to accept calls, for instance we could try to get all the jobs in a given queue: ```powershell $ curl --location 'http://localhost:8080/queues/my-test-queue/jobs' \ --header 'Authorization: Bearer 1234' ``` Will return: ``` Not found% ``` as there are no jobs in any queue yet. Please check the following sections in order to learn how to use all the proxy's features. # Architecture overview BullMQ Proxy is implemented using a relative simple architecture, yet it provides a lot of flexibility. At the core of the proxy we have the [Bun javascript runtime](https://bun.sh/). The choice of this runtime is mostly due to its much better HTTP and WebSocket performance and memory consumption than what is available on other popular runtimes such as NodeJS. The proxy encapsulates BullMQ library (powered by a Redis™, or compatible instance), and provides an HTTP Restful API (and a WebSocket's API soon), that allows any language or framework that supports HTTP clients and servers to interact with the queues as well as to process jobs. For example, adding a bunch of jobs to a queue implies posting an array with jobs to `/queues/:queue-name/jobs`, whereas to process the jobs we need to register an http endpoint (also known as a webhook) that will be called every time there is a job to process. This effectively allows BullMQ to be used on a multi-language multi-framework platform, unlocking easy communication and job management between any services. {% @mermaid/diagram content="graph LR AddJob\[Add Job] -->|HTTP POST /jobs| BullMQProxy\[BullMQ Proxy] BullMQProxy -->|uses| Redis\[(Redis Database)] BullMQProxy -->|HTTP POST /process| Worker1\[Worker 1] BullMQProxy -->|HTTP POST /process| Worker2\[Worker 2] BullMQProxy -->|HTTP POST /process| WorkerN\[Worker N] Worker1 -->|HTTP 200 OK| BullMQProxy Worker2 -->|HTTP 200 OK| BullMQProxy WorkerN -->|HTTP 200 OK| BullMQProxy ``` classDef database fill:#f96,stroke:#333,stroke-width:2px; class Redis database;" %} ``` # Using Dragonfly The proxy also works with Dragonfly, there are a couple of settings that must be taken care of though, the Dragonfly instance must be started using `--cluster_mode=emulated --lock_on_hashtags` and if you do not have enough ram memory you may limit the number of threads with `--proactor_threads=4` (for example, will just limit to max 4 threads). Finally, we need to set a default prefix for all our queues so that we can take advantage of the lock\_on\_hashtags setting (you can read more on why this is needed here) So for example if we where to use docker-compose for runnig the proxy with Dragonfly instead of Redis™ it will look like this: ```yaml version: '3' services: proxy: image: ghcr.io/taskforcesh/bullmq-proxy:latest ports: - 8080:8080 environment: PORT: 8080 REDIS_HOST: redis REDIS_PORT: 6379 REDIS_PASSWORD: ${REDIS_PASSWORD} REDIS_TLS: ${REDIS_TLS} AUTH_TOKENS: ${AUTH_TOKENS} QUEUE_PREFIX: '{b}' dragonflydb: image: docker.dragonflydb.io/dragonflydb/dragonfly environment: DFLY_cluster_mode: emulated DFLY_lock_on_hashtags: true ports: - 6379:6379 ``` # Authentication The Authentication scheme in BullMQ Proxy is currently pretty simple. You add a list of valid comma separated tokens as an env variable (`AUTH_TOKENS`), and then for every request made to the proxy send one of those valid tokens as a BEARER token. ```powershell curl --location 'http://localhost:8080/queues/my-test-queue/jobs' \ --header 'Authorization: Bearer 1234' ``` Sometimes, an endpoint that is processing a job will need to communicate with the proxy, for example for updating a job's progress status or adding logs to the job. This requires a different token, which is actually provided in the body of the call. For example, if the endpoint was implemented in NodeJS it would look something along these lines: ```javascript const http = require('node:http'); // Create an HTTP server const server = http.createServer((req, res) => { const { job, token } = JSON.parse(req.body); // Do something with the job await doSomething(job); // Update progress to 100 const updateProgress = await fetch(`http://localhost:8080/queues/my-test-queue/jobs/${job.id}/progress`, { method: 'POST', body: JSON.stringify(100), headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${token}` } }); res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify('job-result')); }); ``` # Queues Queues are recipients of jobs (also called messages in other systems), a BullMQ Proxy can contain any number of queues, all sharing the same Redis instance. For practical purposes, the number of queues should be kept low, even though they are lightweight and do not consume CPU if they are not being used. Queues can be found on the `/queues` endpoint. There are a number of operations that are provided in this endpoint which we will cover in the next sub pages. # Adding jobs The most basic operation in BullMQ is adding a job. A job is simply a data structure that has a name, some custom data and options. Jobs are added posting a JSON object to the `/queues/:queue-name` endpoint. The endpoint accepts an array of jobs, if sending a standalone job do not forget to wrap it in an array. {% hint style="info" %} Posting jobs is an atomic operation, either all jobs or none will be added to the queue, so if the call fails for any reason, you can be certain that no jobs were added at all. {% endhint %} The JSON object must follow this interface: ```typescript interface Job { name: string; data: any; opts?: JobOpts; } ``` Lets add the simplest possible job: {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'https://myproxy.dev/queues/my-queue/jobs' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer my-secret-token' \ --data '[{ "name": "paint-red", "data": {}, }]' ``` {% endtab %} {% endtabs %} {% hint style="info" %} Note how we wrap the job in an array, as we are only sending one job in this example. {% endhint %} The call should succeed and return the complete Job as a Json object with all the defaults as it is stored in BullMQ: ```json [ { "name": "paint-red", "data": {}, "opts": { "attempts": 0, "delay": 0 }, "id": "116", "progress": 0, "returnvalue": null, "stacktrace": null, "attemptsStarted": 0, "attemptsMade": 0, "delay": 0, "timestamp": 1708597320043, "queueQualifiedName": "bull:my-test-queue" } ] ``` There are obviously many more advanced jobs and features which we will cover in the next pages. # Retries In a normal queue operation it is inevitable that jobs will eventually fail. Failed jobs will by default en on the "failed" set (unless removeOnFailed is enabled). However, it is often useful to perform a number of retries before the job is finally considered failed. In order to enable retries we need to specify a max number of attempts to retry a job, and which backoff strategy to use. These are options passed inside the "opts" field of our job payload following this interface (for simplicity we will hide the rest of the available options for now): ```typescript interface JobOpts { attempts: number; backoff: { type: "exponential" | "linear", delay: number // Time in milliseconds } // ... More opts } ``` We can therefore enable retries by sending these options when adding a job, so for instance if we want to retry up to 5 times with an exponential backoff that starts with 1 second we would send the job like this: {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'http://mydomain.dev/queues/my-queue/jobs' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer my-secret-token' \ --data '[{ "name": "paint-red", "data": {}, "opts": { "attempts": 5, "backoff": { "type": "exponential", "delay": 1000 } }]' ``` {% endtab %} {% endtabs %} Now, if this job would fail when processed, it will retry up to 5 times following an exponential backoff (retrying after 1 second, 2 seconds, 4, seconds and so on). # Delayed jobs Delayed jobs are jobs that will wait the specified amount of time before they are made available to any existing worker. A delayed job is simply added to a queue by specifying the "delay" option, which accepts the number of milliseconds to wait before being processed. {% hint style="info" %} Note that if the workers are busy processing other jobs, the delay job may experience more delay than specified, even though delayed jobs are always placed at the front of the queue bypassing all existing jobs. {% endhint %} Lets expand our job options interface with `delay`: ```typescript interface JobOpts { delay: number; // Milliseconds // ... More opts } ``` {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'http://mydomain.dev/queues/my-queue/jobs' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer my-secret-token' \ --data '[{ "name": "paint-red", "data": {}, "opts": { "delay": 5000 }]' ``` {% endtab %} {% endtabs %} The above snippet will add a job to the queue that will wait 5 seconds before it is attempted by an available worker. # Prioritized Sometimes it is necessary to process jobs following a priority order. This is easily achieved by setting the "priority" option on the job payload. The priorities range from `1` to `2 097 152.` Where the lowest value has the largest priority, following the same schema used by processes in Unix. A job without priority (or priority 0) will have the highest priority and be processed before any other prioritized job. The JobOpts interface with the added priority option would look like this: ```typescript interface JobOpts { priority: number; // an integer between 1 and 2_097_152 // ... More opts } ``` {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'http://mydomain.dev/queues/my-queue/jobs' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer my-secret-token' \ --data '[{ "name": "paint-red", "data": {}, "opts": { "priority": 5 } }, { "name": "paint-black", "data": {} } ]' ``` {% endtab %} {% endtabs %} The above snippet will add one Job with priority 5, and another without priority. The job named "paint-black" even though it was added after the one called "paint-red" will be processed before, as jobs without priority will be processed before jobs with priority. # Repeatable Repeatable jobs are a special type of job that based on a cron expression or some other repetition pattern will keep repeating as long as the conditions are met. Repeatable jobs are a bit tricker to understand as a Repeatable job is composed of 2 elements, first a "meta-job" that keeps the repetition information, secondly the jobs themselves that are created after every iteration. Once a repeatable job has been added to a queue it will keep repeating until either the conditions are not met anymore (such as when the end date has passed), or the repeatable "meta-job" is removed from the queue. {% hint style="danger" %} Repeatable jobs are not yet supported by the Proxy but it is coming soon. {% endhint %} # LIFO LIFO stands for Last-In-First-Out. This mode is the oposite of a standard queue, where jobs are processed in the same order as they arrive to the queue. In a LIFO queue the last added job will be processed before existing jobs. To enable this mode for a given job just pass the "lifo" property and set it to true. ```typescript interface JobOpts { lifo: boolean; // ... More opts } ``` {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'http://mydomain.dev/queues/my-queue/jobs' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer my-secret-token' \ --data '[{ "name": "paint-black", "data": {}, }, { "name": "paint-red", "data": {}, "opts": { "lifo": true } } ]' ``` {% endtab %} {% endtabs %} In the above snippet, the job named "paint-red" will be processed before "paint-black" as it was added with the lifo option set to true. # Custom Job IDs BullMQ automatically generate job ids for all the jobs added to a queue. The job id is an increasing integer number. This is the default and recommended behaviour for most use cases. It is however possible to add jobs with a custom ID that will be used instead of the one auto-generated. The particularity is that there cannot be 2 jobs with the same job ID, so if you add a second job with the same ID as an existing job in the queue, the second job will just be ignored. This is usually useful in order to implement "debouncing". For example, lets say that you have some system that generates a lot of messages, but you only care to process 1 job, at least until that job has been completely processed. By using a custom job ID, only 1 job will exist in the queue until this job has been processed. If you keep the job in the completed set (the default), or the job fails and is kept in the failed set (also the default), then no new jobs will be added to the queue if they use the same job id. Therefore you must choose a proper "removeOnComplete/Fail" strategy suitable for your particular use case. You can read more about these options in the "Workers" section. In order to specify a custom job id, just set the jobId field in the job's options: ```typescript interface JobOpts { jobId: string; // ... More opts } ``` {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'http://mydomain.dev/queues/my-queue/jobs' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer my-secret-token' \ --data '[{ "name": "paint-black", "data": {}, "opts": { "jobId": "my-custom-id" } } ]' ``` {% endtab %} {% endtabs %} This call will add a job with a custom id, note that the data returned by the call now includes the custom id in the id field instead of the autogenerated one. ```json [ { "name": "paint-black", "data": {}, "opts": { "attempts": 0, "delay": 0, "jobId": "my-custom-id" }, "id": "my-custom-id", "progress": 0, "returnvalue": null, "stacktrace": null, "attemptsStarted": 0, "attemptsMade": 0, "delay": 0, "timestamp": 1708618768745, "queueQualifiedName": "bull:my-test-queue" } ] ``` # Getting jobs The proxy also provides convenient endpoints that allow getting jobs, either as a page of jobs based on some filtering options or a given job based on its job ID. A page of jobs follows this interface: ```typescript interface JobsPage { start: number; length: number; count: { waiting: number; active: number; completed: number; failed: number; }, jobs: JsonJob[]; } ``` Access the jobs endpoint providing a start and a length so that you can fetch a particula page of jobs. {% tabs %} {% tab title="Curl" %} ```javascript curl --location 'http://mydomain.dev/queues/my-queue/jobs?\ statuses=completed,failed&start=10&length=20' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer my-secret-token' ``` {% endtab %} {% endtabs %} If you are instead interested in getting one particular job based on its id, just use this endpoint instead: {% tabs %} {% tab title="Curl" %} ``` curl --location 'http://mydomain.dev/queues/my-queue/jobs/my-job-id --header 'Content-Type: application/json' \ --header 'Authorization: Bearer my-secret-token' ``` {% endtab %} {% endtabs %} And if the job exists you will get a [`JobJson`](https://api.docs.bullmq.io/interfaces/v5.JobJson.html) object. # Queue's actions Queues actions allows to perform som admin operations on queues such as pausing, resuming, deleting, and so on. {% hint style="danger" %} This functionality is coming soon... {% endhint %} # Reference Queue API Reference ```typescript interface Job { name: string; data: any; opts?: JobOpts; } ``` ```typescript interface JobOpts { attempts?: number; backoff?: { type: "exponential" | "linear", delay: number // Time in milliseconds }; delay?: number; // Milliseconds priority?: number; // Value between 1 and 2_097_152 lifo?: boolean; jobId?: string; } ``` ```typescript interface JobJson { id: string; data: any; opts: JobOpts; progress?: number | object; returnvalue?: string; failedReason?: string; stacktrace?: string[]; attemptsMade: number; delay: number; timestamp: number; processedOn?: number; finishedOn?: number; queueQualifiedName: string; } ``` ## Add jobs `POST` `/queues/:queue-name/jobs` Adds one or more jobs to a given queue. **Headers** | Name | Value | | ------------- | ------------------ | | Content-Type | `application/json` | | Authorization | `Bearer ` | **Body** **The body must include an array of jobs with the following structure:** | Name | Type | Description | | ------ | ------- | -------------------------------------------- | | `name` | string | Name of the job | | data | any | Arbitrary data to pass to the job processor. | | opts | JobOpts | Job options | **Response** {% tabs %} {% tab title="200" %} ```json [{ "id": string, "name": string, "data": {}, "opts": JobOpts, ... }] ``` {% endtab %} {% tab title="400" %} ```json Error message depending on the cause of the error. ``` {% endtab %} {% endtabs %} ## Get Jobs `GET` `/queues/:queue-name/jobs` Get jobs in a queue in pages and filtered according to their statuses. **Query Params** | Name | Default | Description | | -------- | ------- | ----------------------------------------------------------------------------------------------------- | | statuses | `all` | comma separated lists of valid statuses ("waiting", "delayed", "prioritized", "completed" or "failed" | | start | `0` | offset on where the returned page should start | | length | 10 | Maximum number of job to return in the page | **Headers** | Name | Value | | ------------- | ------------------ | | Content-Type | `application/json` | | Authorization | `Bearer ` | **Response** {% tabs %} {% tab title="200" %} ```json { "counts"; { [status: string]: number }, "jobs": JobJson[], "start": number, "length": number } ``` {% endtab %} {% tab title="400" %} ```json Error message depending on the cause of the error. ``` {% endtab %} {% endtabs %} ## Get Job `GET` `/queues/:queue-name/jobs/job-id` \ **Headers** | Name | Value | | ------------- | ------------------ | | Content-Type | `application/json` | | Authorization | `Bearer ` | **Response** {% tabs %} {% tab title="200" %} ```json JobJson ``` {% endtab %} {% tab title="400" %} ```json { "error": "Invalid request" } ``` {% endtab %} {% endtabs %} # Workers In BullMQ, workers are the things that process jobs. In BullMQ Proxy, the workers are defined as a bunch of options that affects their behaviour as well as an endpoint that will be called by the proxy everytime a job should be processed. The endpoint is what actually performs the job processing. For example, we could have an endpoint connected to a lambda function, so everytime a job needs to be processed, the job data and its options are passed to the lambda and the proxy will wait for the lambda to complete before considering the job to have been completed. The endpoint could obviously also fail, it could timeout or return a status code different from 200 to 209, which would be considered a failure, resulting in the job being moved to the failed set in BullMQ. The workers that are defined in BullMQ Proxy are persisted in Redis™, so that when the proxy restarts, it will remember which workers that should be running. In the next pages we will go through the details of how the workers are used and which options we have at our disposal. # Endpoints Every worker is associated to an endpoint. An endpoint can be seen as a webhook. When we register (add) a worker we are telling the proxy which endpoint should be called for every job that needs to be processed. The endpoint must be a valid url, either http or https, and it must be powerful enough to handle the load that will be generated by BullMQ. In would also be possible to enable rate-limits, so that the endpoint is not called more often than some predefined rate. The endpoint itself may take some time to actually process the job. Here it is important to consider things such as timeouts, which will be handled by the proxy as if the job had failed. Most services have limitations on how long an http request can be, whereas AWS Gateway has a limit of 30 seconds, AWS lambdas have a maximum duration of 15 minutes. Your millage may vary depending on where you deploy the services to be called by the proxy. # Adding workers Workers are added posting a json object to the `/workers` endpoint. An endpoint is specified as an url, which method (http verb) to use on the call to said url, which optional headers to send, as well as an optional timeout, which if reached will consider the call a failure. ```typescript interface WorkerMetadata { queue: string; endpoint: { url: string; method: "post" | "get" | "put" | "delete" | "patch"; headers?: Record; timeout?: number; // max allowed duration in milliseconds } // .. more options } ``` The most basic object we can send in order to start a worker could for example be the following: {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'http://localhost:8080/workers' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer 1234' \ --data '{ "queue": "my-test-queue", "endpoint": { "url": "http://mydomain.dev", "method": "post" } }' ``` {% endtab %} {% endtabs %} The call will return OK -200 if successful, and a worker would now be ready to start processing jobs by calling the given endpoint. When the endpoint is called, it will receive a JSON object with the job including its name, data and options in the body. ```typescript interface Job { name: string; data: any; opts: JobOptions; } ``` The endpoint can do its thing with the job, and when it is ready it must return a status code between 200 and 209 to signal success, and return some optional data that will be stored in the job's `returnvalue` field. If the job should be marked as fail, the call should set a status code different than 200 to 209, and return some error message that will also be stored in the job's `failedReason` field. {% hint style="info" %} Note that only one worker can be defined per queue. If the call is repeated for the same queue, it will just overwrite its options and restart the worker with those new options. {% endhint %} # Concurrency By default, workers will only call endpoints to process jobs one at a time, however it is easy to achieve concurrency by using the concurrency option. This is an optional field that accepts any positive number, as a concurrency factor, i.e. how many calls to an endpoint can be perform in parallel. There is no upper limit on the concurrency factor other than the practical limits of the system running the proxy, Redis™ or the endpoint, please adjust as needed. We can expand our previous interface with a concurrency field: ```typescript interface WorkerMetadata { opts?: { concurrency?: number; // .. more options } // .. more options } ``` As before we just can post this object to the /workers endpoint to register or update an existing endpoint: {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'http://localhost:8080/workers' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer 1234' \ --data '{ "queue": "my-test-queue", "opts": { "concurrency": 50 }, "endpoint": { "url": "http://mydomain.dev", "method": "post" } }' ``` {% endtab %} {% endtabs %} # Rate-Limit A quite popular feature of BullMQ is the possibility to activate a rate-limit on a queue. This rate-limit guarantees that no more jobs than what the rate-limit defines will ever be processed, in the case of the proxy this means that the calls to the endpoint will always be within the rate-limit. Lets add the limiter property to our WorkerMetadata interface: ```typescript interface WorkerMetadata { opts?: { limiter?: { max: number; // integer representing max number of jobs to process on "duration" duration: number: // number of millisecons } // .. more options } // .. more options } ``` The limiter just takes 2 properties, "max" defines the maximum number of jobs that will be processed during "duration", which is some time span defined in milliseconds. So for example, to limit to 100 jobs per second we could just use this object: {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'http://localhost:8080/workers' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer 1234' \ --data '{ "queue": "my-test-queue", "opts": { "concurrency": 300, "limiter": { "max": 100, "duration": 1000 } }, "endpoint": { "url": "http://mydomain.dev", "method": "post" } }' ``` {% endtab %} {% endtabs %} # Removing finished jobs By default, BullMQ will keep all completed and failed jobs. Most likely you will want to limit to sane amounts in order to not fill up your Redis™ database with older jobs. The workers support two options that can be used to specify the behaviour of completed and failed jobs: ```typescript interface WorkerMetadata { opts?: { removeOnComplete?: { count?: number; age?: number; }, removeOnFail?: { count?: number; age?: number; } // .. more options } // .. more options } ``` The "count" field specifies the maximum number of jobs to keep, whereas the "age" specifies the maximum age in seconds of the jobs to keep. You can use any of these options standalone or combined. {% hint style="info" %} Note that the removal of jobs is lazy. So older jobs will only be removed if new jobs are completed or failed. {% endhint %} Lets update our endpoint so that we only keep the latest 1000 completed jobs, and no failed jobs older than one day but never keep more than 5000: ```powershell curl --location 'http://localhost:8080/workers' \ --header 'Content-Type: application/json' \ --header 'Authorization: Bearer 1234' \ --data '{ "queue": "my-test-queue", "opts": { "concurrency": 300, "removeOnComplete": { "count": 1000 }, "removeOnFail": { "count": 5000, "age": 86400 } }, "endpoint": { "url": "http://mydomain.dev", "method": "post" } }' ``` # Stalled jobs In BullMQ, every time is picked by a worker, and during the duration of the processing of the job, a lock, which is represented by a special key in Redis™. The locks express that the worker is alive and the job is being processed normally. If the worker fails to keep this lock from expiring, then the system (other running workers basically) will detect that the job has lost its lock and will mark the job as "stalled". So a stalled job just means that the job has been moved from its active status to either "wait" or "failed" when a lock is missing. The most common scenario where this could happen with BullMQ Proxy is if the proxy is shut down or restarted abruptly without waiting for current processing jobs to complete. The next time the proxy starts and new workers are instantiated they will detect which jobs have been stalled and moved to wait or to failed. {% hint style="info" %} Note that the time it takes to detect a stalled job depends on the stalled checker interval which is by default 30 seconds. {% endhint %} In order to prevent from a worker malfunction to keep stalled jobs from looping between the active status and wait status there is an option (`maxStalledCount` )that restricts the number of times a given job can stall, which is by default **1** (assuming that this is a very rare occurence for a given job). If any job stalls more than "maxStalledCount" the job will instead be moved to failed with the failedReason "job stalled more than allowable limit", indicating that maybe there is something else going on that is making the workers unable to keep the locks alive during the duration of the job. This could happen if for example the jobs are long living and the CPU where the proxy is running approaches 100%, so that the process of renewing locks does not have a chance to run. With this in mind we can expand the WorkerMetadata interface to reflect this option: ```typescript interface WorkerMetadata { opts?: { maxStalledCount?: number; // .. more options } // .. more options } ``` # Timeouts It is possible to define a timeout for the workers so that if the job takes more time than the one specified in this option, the http call to the endpoint will be aborted. The worker will merely request to abort the call, and it will be the endpoint's responsibility to actually gracefully abort whatever operation it may be doing. The job itself will be marked as failed with "failedReason": ... # Removing workers BullMQ Proxy provides an endpoint to remove workers that are not needed anymore. The endpoint is simply `/workers/:queue-name` using "DELETE" as the http verb: {% tabs %} {% tab title="Curl" %} ```powershell curl --location --request DELETE 'http://localhost:8080/workers/my-queue' \ --header 'Authorization: Bearer my-secret-token' ``` {% endtab %} {% endtabs %} If the call succeeds, the worker will be removed from the proxy and will not process new jobs. Note that if the worker is already processing some jobs, the worker will close gracefully to avoid making any jobs stalled. # Getting workers BullMQ Proxy provides a simple api to return all the registered workers in the system. Just call the `/workers` endpoint with the "GET" http verb: {% tabs %} {% tab title="Curl" %} ```powershell curl --location 'http://mydomain.dev/workers/my-queue' \ --header 'Authorization: Bearer my-secret-token' ``` {% endtab %} {% endtabs %} {% hint style="info" %} Currently it only returns the workers that are registered in the system and that **should** be running, however, things like disconnections to Redis™ hosts could prevent the workers from actually process any jobs. Use a BullMQ [dashboard](https://taskforce.sh) tool in order to get a better view of your queues and running workers. {% endhint %} # Reference Workers API Reference ```typescript interface WorkerMetadata { queue: string; endpoint: EndpointOptions; opts?: WorkerOptions; } ``` ```typescript interface EndpointOptions { url: string; method: "post" | "get" | "put" | "delete" | "patch"; headers?: Record; timeout?: number; // max allowed duration in milliseconds } ``` ```typescript interface WorkerOptions { concurrency?: number; limiter?: { max: number; // integer representing max number of jobs to process on "duration" duration: number: // number of millisecons }; removeOnComplete?: { count?: number; age?: number; }; removeOnFail?: { count?: number; age?: number; }; maxStalledCount?: number; } ``` ## Workers API ## Add / Update a worker `POST` `/workers` Adds a new worker or updates an existing one. **Headers** | Name | Value | | ------------- | ------------------ | | Content-Type | `application/json` | | Authorization | `Bearer ` | **Body** | Name | Type | Description | | ---------- | ------------- | ---------------------- | | `queue` | string | Name of the queue | | `endpoint` | Endpoint | Endpoint options | | `opts` | WorkerOptions | Options for the worker | **Response** {% tabs %} {% tab title="200" %} ``` "OK" ``` {% endtab %} {% tab title="400" %} ```json Error message depending on the cause of the error ``` {% endtab %} {% endtabs %} ## Remove worker `DELETE` `/workers/:queue-name` Removes an existing worker for a given queue. **Headers** | Name | Value | | ------------- | ---------------- | | Authorization | `Bearer ` | **Response** {% tabs %} {% tab title="200" %} ```json "OK" ``` {% endtab %} {% tab title="404" %} ``` "Worker not found" ``` {% endtab %} {% endtabs %} ## Get workers `GET` `/workers` Gets all the workers registered in the proxy. **Headers** | Name | Value | | ------------- | ------------------ | | Content-Type | `application/json` | | Authorization | `Bearer ` | **Response** {% tabs %} {% tab title="200" %} ```json { "queue-name": "url" } ``` {% endtab %} {% endtabs %} # Jobs # Jobs' actions There are several actions that can be performed on jobs that are inside a given queue. Specifically there are 2 types of actions: actions that can be performed from the worker when processing a given job, and actions that can be performed while a job is not being processed. When a job is being processed, there are two operations available that can be performed on the job: update its progress and add logs. Both operations are fully optional and their use depends on what kind of jobs we are processing. For example, if it is a heavy job like transcoding a video, we may be interested in updating the jobs progress so that we can check with a queue dashboard or programmatically, the status of a given job. Similarly, if we want to help in debugging a jobs processing we may want to add some logs to the job, so that we can then see exactly what happened during its processing. There are other actions that can be performed on jobs that are not currently being processed, for example, deleting a job, promoting a delayed job so that it is processed sooner than its delay time, manually retrying a failed job among others. These types of actions are normally suitable for dashboards or BullMQ frontends, as they are difficult to apply in practice programmatically. Still, as there are use cases when they are needed, we provide an API for them too. # Update job progress # Add job logs # Reference Reference for jobs' actions. ## Update Job Progress `POST` `/queues/:queue-name/jobs/:job-id/progress` Updates the progress of a given job. The progress can be a number or any other json serializable object. {% hint style="info" %} Note, the token used as bearer token must be the [token](/http-api/authentication) received on the worker's endpoint specific for a given job. {% endhint %} **Headers** | Name | Value | | ------------- | -------------------- | | Content-Type | `application/json` | | Authorization | `Bearer ` | **Body** An integer or any JSON serializable object. **Response** {% tabs %} {% tab title="200" %} ```json OK ``` {% endtab %} {% tab title="400" %} ``` Message with the reason for the invalid request. ``` {% endtab %} {% endtabs %} ## Add Job Logs `POST` `/queues/:queue-name/jobs/:job-id/logs` Add a log to a given job. Any number of logs can be added to a given job. {% hint style="info" %} Note, the token used as bearer token must be the [token](/http-api/authentication) received on the worker's endpoint specific for a given job. {% endhint %} **Headers** | Name | Value | | ------------- | -------------------- | | Content-Type | `application/json` | | Authorization | `Bearer ` | **Body** A plain string with the message to add as a log. **Response** {% tabs %} {% tab title="200" %} ```json OK ``` {% endtab %} {% tab title="400" %} ``` Message with the reason for the error ``` {% endtab %} {% endtabs %} ## Get Job Logs `GET` `/queues/:queue-name/jobs/:job-id/logs?start=0&length=10` \ **Headers** | Name | Value | | ------------- | ------------------ | | Content-Type | `application/json` | | Authorization | `Bearer ` | **Query** | Name | Type | Description | | -------- | ------ | ------------------------ | | `start` | number | First log to return | | `length` | number | Number of logs to return | **Response** {% tabs %} {% tab title="200" %} ```json { "count": number, "logs": string[], } ``` {% endtab %} {% tab title="400" %} ``` Message with the reason for the error ``` {% endtab %} {% endtabs %} # Configuration Environment variables available for BullMQ Proxy The proxy can be configured using env variables that must be set up at the time the service starts in order to have any affect. In the following table we list the available variables and their default values if any.
VariableDefaultDescription
AUTH_TOKENS
Comma separated list of valid tokens that can be used to authenticate in the proxy.
QUEUE_PREFIX
bullPrefix used for all Redis keys used by the queues managed by the proxy.
PORT
8080Port where the proxy will listen for requests.
REDIS_URI
Redis URI string. Either this standalone string or separate Redis™ variables can be used.
REDIS_HOST
localhostHost where the Redis™ instance is running.
REDIS_PORT
6379Port used for the Redis™ instance.
REDIS_USERNAME
Optiona username for the used Redis™ instance.
REDIS_PASSWORD
Password used for the Redis™ instance.
REDIS_TLS
Enable TLS when connecting to the Redis™ instance.
QUEUE_CACHE_SIZE
100Maximum number of BullMQ queue instances to keep in the cache.
DEBUG
falseEnable debugging logs.
MIN_QUEUE_NAME_LENGTH
3Minimum allowed length for queue names.
MAX_QUEUE_NAME_LENGTH
100Maximum allowed length for queue names.
WORKER_METADATA_KEY

bullmq-proxy:workers
Redis™ key where to store metadata for registered workers.
# Debugging It is possible to enable more log messages from the proxy by enabling the `DEBUG` environment variable, this can help in getting more information if something does not work as expected. Another tool that is very helpful when running BullMQ is a dashboard such as [Taskforce.sh](https://taskforce.sh). With this tool it is possible to see all the queues that are available in a Redis™ instance, as well as all the jobs in the different states. This allows to examine the jobs that completed, or maybe more importantly the ones that have failed, so that they can be analyzed and when the root cause resolved optionally retried. There are many more features provided by the dashboard, just give it a try to see what it can do for you.