diff --git a/content/develop/use-cases/job-queue/java-jedis/DemoServer.java b/content/develop/use-cases/job-queue/java-jedis/DemoServer.java index 40667cbee6..4301db7cf2 100644 --- a/content/develop/use-cases/job-queue/java-jedis/DemoServer.java +++ b/content/develop/use-cases/job-queue/java-jedis/DemoServer.java @@ -583,7 +583,7 @@ private static String htmlPage(long visibilityMs) { + "

Redis Job Queue Demo

\n" + "

\n" + " Enqueue background jobs and watch a pool of workers drain them through Redis.\n" - + " Pending jobs sit in a list; each worker uses BRPOPLPUSH to atomically\n" + + " Pending jobs sit in a list; each worker uses BLMOVE to atomically\n" + " claim a job and move it to a processing list. Completed jobs move to a short\n" + " history. If a worker hangs past the " + visibilityMs + " ms visibility timeout,\n" + " the reclaimer moves its job back to pending so no work is lost.\n" diff --git a/content/develop/use-cases/job-queue/java-jedis/RedisJobQueue.java b/content/develop/use-cases/job-queue/java-jedis/RedisJobQueue.java index 3b45182cf9..4fc2496db2 100644 --- a/content/develop/use-cases/job-queue/java-jedis/RedisJobQueue.java +++ b/content/develop/use-cases/job-queue/java-jedis/RedisJobQueue.java @@ -1,6 +1,7 @@ import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; import redis.clients.jedis.Pipeline; +import redis.clients.jedis.args.ListDirection; import redis.clients.jedis.params.ScanParams; import redis.clients.jedis.resps.ScanResult; @@ -219,7 +220,7 @@ public ClaimedJob claim(long timeoutMs) { double timeoutSec = Math.max(timeoutMs / 1000.0, 0.1); String jobId; try (Jedis jedis = pool.getResource()) { - jobId = jedis.brpoplpush(pendingKey, processingKey, (int) Math.ceil(timeoutSec)); + jobId = jedis.blmove(pendingKey, processingKey, ListDirection.RIGHT, ListDirection.LEFT, timeoutSec); } if (jobId == null) { return null; diff --git a/content/develop/use-cases/job-queue/java-jedis/_index.md b/content/develop/use-cases/job-queue/java-jedis/_index.md index 3588ffa785..b576518aa5 100644 --- a/content/develop/use-cases/job-queue/java-jedis/_index.md +++ b/content/develop/use-cases/job-queue/java-jedis/_index.md @@ -38,7 +38,7 @@ The flow looks like this: 1. The application calls `queue.enqueue(payload)` 2. The helper writes the job metadata hash and `LPUSH`es the job ID onto the pending list 3. A worker calls `queue.claim(timeoutMs)` -4. The helper runs `BRPOPLPUSH` to atomically move the next pending ID into the processing list and writes a per-claim `claim_token` plus `claimed_at_ms` on the hash +4. The helper runs `BLMOVE` to atomically move the next pending ID into the processing list and writes a per-claim `claim_token` plus `claimed_at_ms` on the hash 5. The worker runs the job and calls `queue.complete(job, result)` or `queue.fail(job, error)` 6. `complete` removes the job from the processing list, writes the result, and `LPUSH`es the ID onto the completed history (with `LTRIM` and an `EXPIRE` on the hash for cleanup) 7. `fail` either retries the job (back to pending) or moves it to the failed list once retries are exhausted @@ -82,7 +82,7 @@ public class Main { } ``` -Jedis operations are synchronous. The helper acquires a `Jedis` connection per call using `pool.getResource()` inside a try-with-resources block, so connections are returned to the pool even on errors. The blocking `claim()` method holds its own connection for the duration of the `BRPOPLPUSH` wait, which is fine because every other call uses a different connection. +Jedis operations are synchronous. The helper acquires a `Jedis` connection per call using `pool.getResource()` inside a try-with-resources block, so connections are returned to the pool even on errors. The blocking `claim()` method holds its own connection for the duration of the `BLMOVE` wait, which is fine because every other call uses a different connection. ### Data model @@ -116,7 +116,7 @@ queue:jobs-jedis:job:9a4f... The implementation uses: * [`LPUSH`](/content/commands/lpush.md) to add new job IDs to the pending list -* [`BRPOPLPUSH`](/content/commands/brpoplpush.md) to atomically claim a job into the processing list +* [`BLMOVE`](/content/commands/blmove.md) to atomically claim a job into the processing list (the modern replacement for the deprecated [`BRPOPLPUSH`](/content/commands/brpoplpush.md)) * [`LREM`](/content/commands/lrem.md) to remove a claimed job from the processing list on complete or fail * [`LTRIM`](/content/commands/ltrim.md) to cap the completed and failed history lists * [`HSET`](/content/commands/hset.md) / [`HGETALL`](/content/commands/hgetall.md) for job metadata @@ -152,16 +152,16 @@ public String enqueue(Map payload) { The payload is stored as a JSON string so the queue can carry arbitrary nested structures without forcing every field into a hash. -## Claiming jobs with BRPOPLPUSH +## Claiming jobs with BLMOVE -A worker blocks until a job is available, then atomically pops it from the pending list and pushes it onto the processing list. `BRPOPLPUSH` does both in a single Redis call: +A worker blocks until a job is available, then atomically pops it from the pending list and pushes it onto the processing list. `BLMOVE` does both in a single Redis call (it's the modern replacement for `BRPOPLPUSH`, which is deprecated in Redis 6.2+): ```java public ClaimedJob claim(long timeoutMs) { double timeoutSec = Math.max(timeoutMs / 1000.0, 0.1); String jobId; try (Jedis jedis = pool.getResource()) { - jobId = jedis.brpoplpush(pendingKey, processingKey, (int) Math.ceil(timeoutSec)); + jobId = jedis.blmove(pendingKey, processingKey, ListDirection.RIGHT, ListDirection.LEFT, timeoutSec); } if (jobId == null) { return null; @@ -265,7 +265,7 @@ public List reclaimStuck() { } ``` -The Lua script also handles a narrower race: a worker that crashed between `BRPOPLPUSH` and writing `claimed_at_ms`. Those jobs are reclaimed after `2 × visibility_ms` using `enqueued_at_ms` as a fallback timer, so they aren't stranded forever. +The Lua script also handles a narrower race: a worker that crashed between `BLMOVE` and writing `claimed_at_ms`. Those jobs are reclaimed after `2 × visibility_ms` using `enqueued_at_ms` as a fallback timer, so they aren't stranded forever. ## Stats and history @@ -306,7 +306,7 @@ The completed and failed lists are capped via `LTRIM` so they never grow unbound Before running the demo, make sure that: -* Redis 6.2 or later is running locally on the default port (6379). Earlier versions still work, since the helper uses commands that have existed since Redis 2.6. +* Redis 6.2 or later running locally on the default port (6379). `BLMOVE` requires Redis 6.2+; on older servers, replace the call with `BRPOPLPUSH`. * Java 17 or later (the demo uses text-block-free string concatenation but still relies on a modern JDK). * Jedis 5.x is on the classpath. The smallest workable classpath is the Jedis jar plus its two transitive dependencies, `commons-pool2` and `slf4j-api`. @@ -393,7 +393,7 @@ The demo only reclaims when you click the button. In production, run `reclaimStu ### Size `JedisPool` for your worker count -`JedisPool` is thread-safe and connections are released back to the pool when the try-with-resources block exits. The demo bumps `maxTotal` to 32 to support the blocking `BRPOPLPUSH` call held by each worker plus the per-request connections used by the HTTP handlers. As a rule of thumb, `maxTotal` should be at least *workers + concurrent HTTP request threads + reclaimer threads + headroom*. +`JedisPool` is thread-safe and connections are released back to the pool when the try-with-resources block exits. The demo bumps `maxTotal` to 32 to support the blocking `BLMOVE` call held by each worker plus the per-request connections used by the HTTP handlers. As a rule of thumb, `maxTotal` should be at least *workers + concurrent HTTP request threads + reclaimer threads + headroom*. ### Use a separate Redis database or key prefix per queue @@ -433,7 +433,7 @@ redis-cli --scan --pattern 'queue:jobs-jedis:*' | xargs redis-cli DEL This example uses the following Redis commands: * [`LPUSH`](/content/commands/lpush.md) to enqueue a job ID. -* [`BRPOPLPUSH`](/content/commands/brpoplpush.md) to atomically claim a job into the processing list. +* [`BLMOVE`](/content/commands/blmove.md) to atomically claim a job into the processing list. * [`LREM`](/content/commands/lrem.md) to remove a job from the processing list on complete or fail. * [`LRANGE`](/content/commands/lrange.md) and [`LLEN`](/content/commands/llen.md) to read queue depth and list contents. * [`LTRIM`](/content/commands/ltrim.md) to cap the completed and failed history. diff --git a/content/develop/use-cases/job-queue/php/JobQueue.php b/content/develop/use-cases/job-queue/php/JobQueue.php index a214013ec6..1a256584b7 100644 --- a/content/develop/use-cases/job-queue/php/JobQueue.php +++ b/content/develop/use-cases/job-queue/php/JobQueue.php @@ -116,7 +116,7 @@ class JobQueue // and moves any job past the visibility timeout back to the pending // list. A job is past the timeout if either: // - claimed_at_ms is set and (now - claimed_at_ms) > visibility_ms, OR - // - claimed_at_ms is missing (worker crashed between BRPOPLPUSH and + // - claimed_at_ms is missing (worker crashed between BLMOVE and // the metadata write) and (now - enqueued_at_ms) > 2 * visibility_ms. // Runs in one round trip so a concurrent worker can't claim a // half-reclaimed job. @@ -266,17 +266,16 @@ public function enqueue(array $payload): string /** * Block until a job is available, then atomically claim it. * - * Uses BRPOPLPUSH to wait for a pending job and move it to the + * Uses BLMOVE to wait for a pending job and move it to the * processing list in a single Redis call. Returns null if nothing * arrives before $timeoutMs. */ public function claim(int $timeoutMs = 1000): ?ClaimedJob { - // Predis brpoplpush takes whole seconds (server-side blocking - // semantics). Round up so a sub-second timeout still blocks at - // least once. - $timeoutSec = max(1, (int) ceil($timeoutMs / 1000)); - $jobId = $this->redis->brpoplpush($this->pendingKey, $this->processingKey, $timeoutSec); + // BLMOVE takes a fractional timeout in seconds. Keep a 0.1s minimum so + // a tiny timeoutMs does not turn into 0, which would block forever. + $timeoutSec = max($timeoutMs / 1000.0, 0.1); + $jobId = $this->redis->blmove($this->pendingKey, $this->processingKey, 'RIGHT', 'LEFT', $timeoutSec); if ($jobId === null || $jobId === false || $jobId === '') { return null; } diff --git a/content/develop/use-cases/job-queue/php/_index.md b/content/develop/use-cases/job-queue/php/_index.md index 98cfd7f3aa..42172f9e8f 100644 --- a/content/develop/use-cases/job-queue/php/_index.md +++ b/content/develop/use-cases/job-queue/php/_index.md @@ -37,7 +37,7 @@ The flow looks like this: 1. The application calls `$queue->enqueue($payload)` 2. The helper writes the job metadata hash and `LPUSH`es the job ID onto the pending list 3. A worker calls `$queue->claim($timeoutMs)` -4. The helper runs `BRPOPLPUSH` to atomically move the next pending ID into the processing list and writes a per-claim `claim_token` plus `claimed_at_ms` on the hash +4. The helper runs `BLMOVE` to atomically move the next pending ID into the processing list and writes a per-claim `claim_token` plus `claimed_at_ms` on the hash 5. The worker runs the job and calls `$queue->complete($job, $result)` or `$queue->fail($job, $error)` 6. `complete` removes the job from the processing list, writes the result, and `LPUSH`es the ID onto the completed history (with `LTRIM` and an `EXPIRE` on the hash for cleanup) 7. `fail` either retries the job (back to pending) or moves it to the failed list once retries are exhausted @@ -110,7 +110,7 @@ Because PHP's built-in dev server runs each HTTP request in a fresh process, per The implementation uses: * [`LPUSH`](/content/commands/lpush.md) to add new job IDs to the pending list -* [`BRPOPLPUSH`](/content/commands/brpoplpush.md) to atomically claim a job into the processing list +* [`BLMOVE`](/content/commands/blmove.md) to atomically claim a job into the processing list (the modern replacement for the deprecated [`BRPOPLPUSH`](/content/commands/brpoplpush.md)) * [`LREM`](/content/commands/lrem.md) to remove a claimed job from the processing list on complete or fail * [`LTRIM`](/content/commands/ltrim.md) to cap the completed and failed history lists * [`HSET`](/content/commands/hset.md) / [`HGETALL`](/content/commands/hgetall.md) for job metadata @@ -148,15 +148,15 @@ public function enqueue(array $payload): string The payload is stored as JSON so the queue can carry arbitrary nested structures without forcing every field into a hash. The `flattenFields()` helper turns the associative `$meta` array into the variadic `field, value, field, value` argument list that Predis's `hset()` expects in Predis 3.x. -## Claiming jobs with BRPOPLPUSH +## Claiming jobs with BLMOVE -A worker blocks until a job is available, then atomically pops it from the pending list and pushes it onto the processing list. `BRPOPLPUSH` does both in a single Redis call: +A worker blocks until a job is available, then atomically pops it from the pending list and pushes it onto the processing list. `BLMOVE` does both in a single Redis call (it's the modern replacement for `BRPOPLPUSH`, which is deprecated in Redis 6.2+): ```php public function claim(int $timeoutMs = 1000): ?ClaimedJob { - $timeoutSec = max(1, (int) ceil($timeoutMs / 1000)); - $jobId = $this->redis->brpoplpush($this->pendingKey, $this->processingKey, $timeoutSec); + $timeoutSec = max($timeoutMs / 1000.0, 0.1); + $jobId = $this->redis->blmove($this->pendingKey, $this->processingKey, 'RIGHT', 'LEFT', $timeoutSec); if ($jobId === null || $jobId === false || $jobId === '') { return null; } @@ -186,7 +186,7 @@ public function claim(int $timeoutMs = 1000): ?ClaimedJob The `claim_token` is the worker's proof of ownership for this attempt. Every subsequent state change (complete, fail) checks it before touching the processing list, so a worker that hung past the visibility timeout cannot interfere with the new claimant. -Predis exposes `BRPOPLPUSH` directly and accepts a whole-second timeout; sub-second blocking would need either a custom command or a non-blocking poll loop. +Predis exposes `BLMOVE` directly and accepts a fractional timeout in seconds, so the demo keeps a 0.1 second minimum instead of rounding up to a whole second. ## Completing jobs @@ -270,7 +270,7 @@ public function reclaimStuck(): array } ``` -The Lua script also handles a narrower race: a worker that crashed between `BRPOPLPUSH` and writing `claimed_at_ms`. Those jobs are reclaimed after `2 × visibility_ms` using `enqueued_at_ms` as a fallback timer, so they aren't stranded forever. +The Lua script also handles a narrower race: a worker that crashed between `BLMOVE` and writing `claimed_at_ms`. Those jobs are reclaimed after `2 × visibility_ms` using `enqueued_at_ms` as a fallback timer, so they aren't stranded forever. ## Stats and history @@ -305,7 +305,7 @@ The completed and failed lists are capped via `LTRIM` so they never grow unbound ## Prerequisites -* Redis 6.2 or later running locally on the default port (6379). Earlier versions still work, since the helper uses commands that have existed since Redis 2.6. +* Redis 6.2 or later running locally on the default port (6379). `BLMOVE` requires Redis 6.2+; on older servers, replace the call with `BRPOPLPUSH`. * PHP 8.1 or later, with the `posix` and `pcntl` extensions enabled (both ship with the official PHP binary on macOS and most Linux distros). * The Predis client (3.x). Install it with [Composer](https://getcomposer.org/): @@ -450,7 +450,7 @@ redis-cli --scan --pattern 'queue:jobs:*' | xargs redis-cli DEL This example uses the following Redis commands: * [`LPUSH`](/content/commands/lpush.md) to enqueue a job ID. -* [`BRPOPLPUSH`](/content/commands/brpoplpush.md) to atomically claim a job into the processing list. +* [`BLMOVE`](/content/commands/blmove.md) to atomically claim a job into the processing list. * [`LREM`](/content/commands/lrem.md) to remove a job from the processing list on complete or fail. * [`LRANGE`](/content/commands/lrange.md) and [`LLEN`](/content/commands/llen.md) to read queue depth and list contents. * [`LTRIM`](/content/commands/ltrim.md) to cap the completed and failed history. diff --git a/content/develop/use-cases/job-queue/php/demo_server.php b/content/develop/use-cases/job-queue/php/demo_server.php index 5e85d32646..bae861b0fa 100644 --- a/content/develop/use-cases/job-queue/php/demo_server.php +++ b/content/develop/use-cases/job-queue/php/demo_server.php @@ -360,7 +360,7 @@ function render_page(int $visibilityMs): string

Redis Job Queue Demo

Enqueue background jobs and watch a pool of workers drain them through Redis. - Pending jobs sit in a list; each worker uses BRPOPLPUSH to atomically + Pending jobs sit in a list; each worker uses BLMOVE to atomically claim a job and move it to a processing list. Completed jobs move to a short history. If a worker hangs past the {$visibilityMs} ms visibility timeout, the reclaimer moves its job back to pending so no work is lost. diff --git a/content/develop/use-cases/job-queue/redis-py/_index.md b/content/develop/use-cases/job-queue/redis-py/_index.md index 2e82082c32..f32eb6e4e7 100644 --- a/content/develop/use-cases/job-queue/redis-py/_index.md +++ b/content/develop/use-cases/job-queue/redis-py/_index.md @@ -37,7 +37,7 @@ The flow looks like this: 1. The application calls `queue.enqueue(payload)` 2. The helper writes the job metadata hash and `LPUSH`es the job ID onto the pending list 3. A worker calls `queue.claim(timeout_ms)` -4. The helper runs `BRPOPLPUSH` to atomically move the next pending ID into the processing list and writes a per-claim `claim_token` plus `claimed_at_ms` on the hash +4. The helper runs `BLMOVE` to atomically move the next pending ID into the processing list and writes a per-claim `claim_token` plus `claimed_at_ms` on the hash 5. The worker runs the job and calls `queue.complete(job, result)` or `queue.fail(job, error)` 6. `complete` removes the job from the processing list, writes the result, and `LPUSH`es the ID onto the completed history (with `LTRIM` and an `EXPIRE` on the hash for cleanup) 7. `fail` either retries the job (back to pending) or moves it to the failed list once retries are exhausted @@ -105,7 +105,7 @@ queue:jobs:job:9a4f... The implementation uses: * [`LPUSH`](/content/commands/lpush.md) to add new job IDs to the pending list -* [`BRPOPLPUSH`](/content/commands/brpoplpush.md) to atomically claim a job into the processing list +* [`BLMOVE`](/content/commands/blmove.md) to atomically claim a job into the processing list (the modern replacement for the deprecated [`BRPOPLPUSH`](/content/commands/brpoplpush.md)) * [`LREM`](/content/commands/lrem.md) to remove a claimed job from the processing list on complete or fail * [`LTRIM`](/content/commands/ltrim.md) to cap the completed and failed history lists * [`HSET`](/content/commands/hset.md) / [`HGETALL`](/content/commands/hgetall.md) for job metadata @@ -138,14 +138,14 @@ def enqueue(self, payload: dict) -> str: The payload is stored as JSON so the queue can carry arbitrary nested structures without forcing every field into a hash. -## Claiming jobs with BRPOPLPUSH +## Claiming jobs with BLMOVE -A worker blocks until a job is available, then atomically pops it from the pending list and pushes it onto the processing list. `BRPOPLPUSH` does both in a single Redis call: +A worker blocks until a job is available, then atomically pops it from the pending list and pushes it onto the processing list. `BLMOVE` does both in a single Redis call (it's the modern replacement for `BRPOPLPUSH`, which is deprecated in Redis 6.2+): ```python def claim(self, timeout_ms: int = 1000) -> Optional[ClaimedJob]: timeout_s = max(timeout_ms / 1000.0, 0.1) - job_id = self.redis.brpoplpush(self.pending_key, self.processing_key, timeout=timeout_s) + job_id = self.redis.blmove(self.pending_key, self.processing_key, timeout_s, src="RIGHT", dest="LEFT") if job_id is None: return None @@ -229,7 +229,7 @@ def reclaim_stuck(self) -> list[str]: return list(reclaimed) ``` -The Lua script also handles a narrower race: a worker that crashed between `BRPOPLPUSH` and writing `claimed_at_ms`. Those jobs are reclaimed after `2 × visibility_ms` using `enqueued_at_ms` as a fallback timer, so they aren't stranded forever. +The Lua script also handles a narrower race: a worker that crashed between `BLMOVE` and writing `claimed_at_ms`. Those jobs are reclaimed after `2 × visibility_ms` using `enqueued_at_ms` as a fallback timer, so they aren't stranded forever. ## Stats and history @@ -260,7 +260,7 @@ The completed and failed lists are capped via `LTRIM` so they never grow unbound ## Prerequisites -* Redis 6.2 or later running locally on the default port (6379). Earlier versions still work, since the helper uses commands that have existed since Redis 2.6. +* Redis 6.2 or later running locally on the default port (6379). `BLMOVE` requires Redis 6.2+; on older servers, replace the call with `BRPOPLPUSH`. * Python 3.9 or later. * The `redis-py` client. Install it with: @@ -365,7 +365,7 @@ redis-cli --scan --pattern 'queue:jobs:*' | xargs redis-cli DEL This example uses the following Redis commands: * [`LPUSH`](/content/commands/lpush.md) to enqueue a job ID. -* [`BRPOPLPUSH`](/content/commands/brpoplpush.md) to atomically claim a job into the processing list. +* [`BLMOVE`](/content/commands/blmove.md) to atomically claim a job into the processing list. * [`LREM`](/content/commands/lrem.md) to remove a job from the processing list on complete or fail. * [`LRANGE`](/content/commands/lrange.md) and [`LLEN`](/content/commands/llen.md) to read queue depth and list contents. * [`LTRIM`](/content/commands/ltrim.md) to cap the completed and failed history. diff --git a/content/develop/use-cases/job-queue/redis-py/demo_server.py b/content/develop/use-cases/job-queue/redis-py/demo_server.py index e31ae91193..7cbb0494b6 100644 --- a/content/develop/use-cases/job-queue/redis-py/demo_server.py +++ b/content/develop/use-cases/job-queue/redis-py/demo_server.py @@ -358,7 +358,7 @@ def _html_page(self) -> str:

Redis Job Queue Demo

Enqueue background jobs and watch a pool of workers drain them through Redis. - Pending jobs sit in a list; each worker uses BRPOPLPUSH to atomically + Pending jobs sit in a list; each worker uses BLMOVE to atomically claim a job and move it to a processing list. Completed jobs move to a short history. If a worker hangs past the {visibility_ms} ms visibility timeout, the reclaimer moves its job back to pending so no work is lost. diff --git a/content/develop/use-cases/job-queue/redis-py/job_queue.py b/content/develop/use-cases/job-queue/redis-py/job_queue.py index 0f257f8e57..1e690d219d 100644 --- a/content/develop/use-cases/job-queue/redis-py/job_queue.py +++ b/content/develop/use-cases/job-queue/redis-py/job_queue.py @@ -79,7 +79,7 @@ # moves any job past the visibility timeout back to the pending list. # A job is past the timeout if either: # - claimed_at_ms is set and (now - claimed_at_ms) > visibility_ms, OR -# - claimed_at_ms is missing (worker crashed between BRPOPLPUSH and the +# - claimed_at_ms is missing (worker crashed between BLMOVE and the # metadata write) and (now - enqueued_at_ms) > 2 * visibility_ms. # Runs in one round trip so a concurrent worker can't claim a # half-reclaimed job. @@ -201,13 +201,13 @@ def enqueue(self, payload: dict) -> str: def claim(self, timeout_ms: int = 1000) -> Optional[ClaimedJob]: """Block until a job is available, then atomically claim it. - Uses ``BRPOPLPUSH`` to wait for a pending job and move it to the + Uses ``BLMOVE`` to wait for a pending job and move it to the processing list in a single Redis call; falls back to the same operation via a Lua script if a job arrives during the blocking window. Returns ``None`` if nothing arrives before ``timeout_ms``. """ timeout_s = max(timeout_ms / 1000.0, 0.1) - job_id = self.redis.brpoplpush(self.pending_key, self.processing_key, timeout=timeout_s) + job_id = self.redis.blmove(self.pending_key, self.processing_key, timeout_s, src="RIGHT", dest="LEFT") if job_id is None: return None diff --git a/content/develop/use-cases/job-queue/ruby/_index.md b/content/develop/use-cases/job-queue/ruby/_index.md index 17cc59d2ac..eedfc046c2 100644 --- a/content/develop/use-cases/job-queue/ruby/_index.md +++ b/content/develop/use-cases/job-queue/ruby/_index.md @@ -37,7 +37,7 @@ The flow looks like this: 1. The application calls `queue.enqueue(payload)`. 2. The helper writes the job metadata hash and `LPUSH`es the job ID onto the pending list. 3. A worker calls `queue.claim(timeout_ms: ...)`. -4. The helper runs `BRPOPLPUSH` to atomically move the next pending ID into the processing list and writes a per-claim `claim_token` plus `claimed_at_ms` on the hash. +4. The helper runs `BLMOVE` to atomically move the next pending ID into the processing list and writes a per-claim `claim_token` plus `claimed_at_ms` on the hash. 5. The worker runs the job and calls `queue.complete(job, result)` or `queue.fail(job, error)`. 6. `complete` removes the job from the processing list, writes the result, and `LPUSH`es the ID onto the completed history (with `LTRIM` and an `EXPIRE` on the hash for cleanup). 7. `fail` either retries the job (back to pending) or moves it to the failed list once retries are exhausted. @@ -108,7 +108,7 @@ queue:jobs:job:9a4f... The implementation uses: * [`LPUSH`](/content/commands/lpush.md) to add new job IDs to the pending list. -* [`BRPOPLPUSH`](/content/commands/brpoplpush.md) to atomically claim a job into the processing list. +* [`BLMOVE`](/content/commands/blmove.md) to atomically claim a job into the processing list (the modern replacement for the deprecated [`BRPOPLPUSH`](/content/commands/brpoplpush.md)). * [`LREM`](/content/commands/lrem.md) to remove a claimed job from the processing list on complete or fail. * [`LTRIM`](/content/commands/ltrim.md) to cap the completed and failed history lists. * [`HSET`](/content/commands/hset.md) / [`HGETALL`](/content/commands/hgetall.md) for job metadata. @@ -144,14 +144,14 @@ end The payload is stored as JSON so the queue can carry arbitrary nested structures without forcing every field into a hash. -## Claiming jobs with BRPOPLPUSH +## Claiming jobs with BLMOVE -A worker blocks until a job is available, then atomically pops it from the pending list and pushes it onto the processing list. `BRPOPLPUSH` does both in a single Redis call: +A worker blocks until a job is available, then atomically pops it from the pending list and pushes it onto the processing list. `BLMOVE` does both in a single Redis call (it's the modern replacement for `BRPOPLPUSH`, which is deprecated in Redis 6.2+): ```ruby def claim(timeout_ms: 1000) timeout_s = [timeout_ms / 1000.0, 0.1].max - job_id = @redis.brpoplpush(@pending_key, @processing_key, timeout: timeout_s) + job_id = @redis.blmove(@pending_key, @processing_key, "RIGHT", "LEFT", timeout: timeout_s) return nil if job_id.nil? token = SecureRandom.hex(8) @@ -248,7 +248,7 @@ def reclaim_stuck end ``` -The Lua script also handles a narrower race: a worker that crashed between `BRPOPLPUSH` and writing `claimed_at_ms`. Those jobs are reclaimed after `2 × visibility_ms` using `enqueued_at_ms` as a fallback timer, so they aren't stranded forever. +The Lua script also handles a narrower race: a worker that crashed between `BLMOVE` and writing `claimed_at_ms`. Those jobs are reclaimed after `2 × visibility_ms` using `enqueued_at_ms` as a fallback timer, so they aren't stranded forever. ## Stats and history @@ -281,7 +281,7 @@ Totals live in a Redis hash so each worker thread — which holds its own Redis ## Prerequisites -* Redis 6.2 or later running locally on the default port (6379). Earlier versions still work, since the helper uses commands that have existed since Redis 2.6. +* Redis 6.2 or later running locally on the default port (6379). `BLMOVE` requires Redis 6.2+; on older servers, replace the call with `BRPOPLPUSH`. * Ruby 3.0 or later. * The `redis-rb` client (5.x) and the `webrick` gem (a stdlib gem in Ruby 3 — install it explicitly with `gem install webrick` if it isn't already on your load path): @@ -338,7 +338,7 @@ The demo includes a small `Worker` and `WorkerPool` ([source](https://github.com The `fail_rate` and `hang_rate` knobs let you watch the at-least-once delivery and reclaim behaviours from the UI without writing test code. -Each `Worker` runs in its own `Thread` and holds its **own** `RedisJobQueue` instance backed by a dedicated Redis connection. A blocking `BRPOPLPUSH` reserves the underlying connection until it returns, so giving each worker its own connection keeps the HTTP server's stats and jobs endpoints responsive even while every worker is parked on a claim call. The `WorkerPool` is constructed with a `queue_factory` lambda so it can mint one fresh helper per worker. +Each `Worker` runs in its own `Thread` and holds its **own** `RedisJobQueue` instance backed by a dedicated Redis connection. A blocking `BLMOVE` reserves the underlying connection until it returns, so giving each worker its own connection keeps the HTTP server's stats and jobs endpoints responsive even while every worker is parked on a claim call. The `WorkerPool` is constructed with a `queue_factory` lambda so it can mint one fresh helper per worker. ## Production usage @@ -364,7 +364,7 @@ A blanket `max_attempts = 3` is a reasonable default for transient failures (net ### Give each blocking worker its own connection -`redis-rb` 5.x is thread-safe — every call through a `Redis` instance is serialised on an internal mutex — but a `BRPOPLPUSH` parks the connection until the call returns. If multiple Ruby threads share a single `Redis` instance and one is blocked on a claim, the others wait behind it. In the demo, the `WorkerPool` builds a fresh `Redis` (and a fresh `RedisJobQueue`) per worker. In production, use a connection pool (such as the `connection_pool` gem) and check out a dedicated connection for blocking commands. +`redis-rb` 5.x is thread-safe — every call through a `Redis` instance is serialised on an internal mutex — but a `BLMOVE` parks the connection until the call returns. If multiple Ruby threads share a single `Redis` instance and one is blocked on a claim, the others wait behind it. In the demo, the `WorkerPool` builds a fresh `Redis` (and a fresh `RedisJobQueue`) per worker. In production, use a connection pool (such as the `connection_pool` gem) and check out a dedicated connection for blocking commands. ### Inspect queue state directly in Redis @@ -395,7 +395,7 @@ redis-cli --scan --pattern 'queue:jobs:*' | xargs redis-cli DEL This example uses the following Redis commands: * [`LPUSH`](/content/commands/lpush.md) to enqueue a job ID. -* [`BRPOPLPUSH`](/content/commands/brpoplpush.md) to atomically claim a job into the processing list. +* [`BLMOVE`](/content/commands/blmove.md) to atomically claim a job into the processing list. * [`LREM`](/content/commands/lrem.md) to remove a job from the processing list on complete or fail. * [`LRANGE`](/content/commands/lrange.md) and [`LLEN`](/content/commands/llen.md) to read queue depth and list contents. * [`LTRIM`](/content/commands/ltrim.md) to cap the completed and failed history. diff --git a/content/develop/use-cases/job-queue/ruby/demo_server.rb b/content/develop/use-cases/job-queue/ruby/demo_server.rb index 903aea34f1..176f6f807a 100644 --- a/content/develop/use-cases/job-queue/ruby/demo_server.rb +++ b/content/develop/use-cases/job-queue/ruby/demo_server.rb @@ -224,7 +224,7 @@ def html_page(visibility_ms)

Redis Job Queue Demo

Enqueue background jobs and watch a pool of workers drain them through Redis. - Pending jobs sit in a list; each worker uses BRPOPLPUSH to atomically + Pending jobs sit in a list; each worker uses BLMOVE to atomically claim a job and move it to a processing list. Completed jobs move to a short history. If a worker hangs past the #{visibility_ms} ms visibility timeout, the reclaimer moves its job back to pending so no work is lost. diff --git a/content/develop/use-cases/job-queue/ruby/job_queue.rb b/content/develop/use-cases/job-queue/ruby/job_queue.rb index 06b8f305b8..91e1ad2896 100644 --- a/content/develop/use-cases/job-queue/ruby/job_queue.rb +++ b/content/develop/use-cases/job-queue/ruby/job_queue.rb @@ -71,7 +71,7 @@ # moves any job past the visibility timeout back to the pending list. # A job is past the timeout if either: # - claimed_at_ms is set and (now - claimed_at_ms) > visibility_ms, OR -# - claimed_at_ms is missing (worker crashed between BRPOPLPUSH and the +# - claimed_at_ms is missing (worker crashed between BLMOVE and the # metadata write) and (now - enqueued_at_ms) > 2 * visibility_ms. # Runs in one round trip so a concurrent worker can't claim a # half-reclaimed job. @@ -159,12 +159,12 @@ def enqueue(payload) # Block until a job is available, then atomically claim it. # - # Uses BRPOPLPUSH to wait for a pending job and move it to the + # Uses BLMOVE to wait for a pending job and move it to the # processing list in a single Redis call. Returns nil if nothing arrives # before timeout_ms. def claim(timeout_ms: 1000) timeout_s = [timeout_ms / 1000.0, 0.1].max - job_id = @redis.brpoplpush(@pending_key, @processing_key, timeout: timeout_s) + job_id = @redis.blmove(@pending_key, @processing_key, "RIGHT", "LEFT", timeout: timeout_s) return nil if job_id.nil? token = SecureRandom.hex(8) diff --git a/content/develop/use-cases/job-queue/ruby/worker.rb b/content/develop/use-cases/job-queue/ruby/worker.rb index 60c4ec076b..eab0d7f4d0 100644 --- a/content/develop/use-cases/job-queue/ruby/worker.rb +++ b/content/develop/use-cases/job-queue/ruby/worker.rb @@ -106,7 +106,7 @@ def pick_outcome # A pool of named Worker threads that can be started and stopped. # # Each worker gets its own RedisJobQueue (built from queue_factory) so its -# blocking BRPOPLPUSH does not stall any other thread sharing the +# blocking BLMOVE does not stall any other thread sharing the # connection. class WorkerPool attr_reader :work_latency_ms, :fail_rate, :hang_rate