Repository navigation
-
Fix:
shoryuken sqs dump/mvno longer stop early on real SQS (mensfeld)find_allused short polling and broke on the first emptyreceive_messageresponse. Real
(distributed) SQS routinely returns an empty batch while the queue still has messages, so dump/mv
quietly processed only a fraction of the queue (invisible on single-node ElasticMQ/LocalStack)- It now long-polls and only stops after several consecutive empty batches, so the queue is actually drained
- Because a longer drain can outlast the queue's visibility timeout (dump/mv delete only after
find_all
returns), re-read messages are now deduplicated by message id, so they are not dumped/moved or counted
twice, and the newest receipt handle is kept so the deferred delete still succeeds - Deletion stays deferred (a failed dump/mv deletes nothing), so a queue larger than SQS's in-flight limit
(~120k standard / ~20k FIFO) drains over multiple runs rather than one - no message is lost
-
Fix: Graceful stop no longer deadlocks when a manager's dispatch loop never started (mensfeld)
await_dispatching_in_progressblocks on aQueuethat is only closed from insidedispatch_loopwhen it
observes the stop flag. If the loop never runs - a graceful stop arriving before the startFutureis
scheduled, or an embedded host whose executor pool is saturated somanager.startnever runs - the signal
was never closed andLauncher#stopdeadlockedstop_new_dispatchingnow closes the release signal itself when the loop hasn't started (tracked via a new
flag);Queue#closeis idempotent, so a loop that does start later and closes it again is harmless
-
Feature:
Shoryuken.fifo_message_deduplicationto opt out of content-based dedup id generation for raw sends (mensfeld)Queue#add_fifo_attributes!always setmessage_deduplication_idto a SHA256 of the body when none was
given, so two raw sends of an identical body within SQS's 5-minute window (Worker.perform_async,
Queue#send_message/#send_messages) were silently deduplicated - the second dropped. ActiveJob got an
opt-out in #1017 but the raw path had none- Set
Shoryuken.fifo_message_deduplication = falseto stop auto-generating that id, so identical bodies are
no longer silently dropped (provide amessage_deduplication_idyourself or enable
ContentBasedDeduplicationon the queue) - Defaults to
true, preserving the existing behavior; an explicitmessage_deduplication_idis still honored
-
Fix: Harden
ExponentialBackoffRetryvisibility handling (mensfeld)next_visibility_timeoutcould return a negative value when a job ran past the 12h SQS ceiling (its
max_timeoutgoes below zero), andchange_message_visibilityrejects a negative timeout; it now clamps
to 0 so the message is retried as soon as possiblehandle_failurecalledchange_visibilityunguarded, so a failure there (e.g. an expired receipt handle)
escaped therescuein#calland masked the original worker error from exception handlers/notifiers; it
now rescues, logs, and reports "not retried" so the original error is re-raised and the message falls back
to the queue's default visibility timeout- The failure
rescuein#callis now scoped to the single-messageyieldonly, so a batch worker's error
no longer routes the message Array intohandle_failure(raising aNoMethodErrorthat masked the original)
-
Fix:
perform_asyncno longer mutates the caller-supplied options hash (mensfeld)DefaultExecutor#perform_asyncandInlineExecutor#perform_asyncdeleted:queue, injected
:message_body, and wroteshoryuken_classinto the nested:message_attributesin place, so a caller
reusing one options hash across enqueues had:queuestripped after the first call - silently routing later
jobs to the worker's default queue- Both now operate on a
dupand rebuild:message_attributeswithmerge, leaving the caller's hash untouched
-
Fix:
Shoryuken::Client.queuesno longer builds the same queue more than once under concurrency (mensfeld)- The cache used an unsynchronized
@@queues[name] ||= Shoryuken::Queue.new(...). Building a queue makes
SQS API calls, and that I/O releases the GVL, so concurrent first-access (dispatch, processor-completion
and worker threads all call it) built the queue multiple times - redundant API calls, and a corrupt cache
on JRuby/TruffleRuby - Access to the cache is now guarded by a mutex
- The cache used an unsynchronized
-
Fix: Busy-processor counter no longer leaks when the executor rejects a worker post (mensfeld)
Manager#assignincrements@busy_processorsbefore posting the workerConcurrent::Promise, but the
matching decrement (processor_done) runs inside the promise body. When the post is rejected with
Concurrent::RejectedExecutionError- a hard stop racing therunning?check, or a saturated bounded custom
launcher_executor- the body never runs and the counter leaks- With a bounded executor the leak is permanent:
ready(@max_processors - busy) keeps shrinking until
dispatch stalls and the group silently stops processing - The increment is now rolled back on
RejectedExecutionErrorby decrementing directly (the message was never
processed, so the FIFOmessage_processedcallback must not run) (#1029)