Tuber

The Ruby client for Tuber, the simple, fast job queue server. One binary, zero dependencies — with unique/idempotent jobs, concurrency keys, job group pipelines, weighted tubes and batch operations built in.

Tuber (the server and this gem) is wire-compatible with beanstalkd: point this client at a stock beanstalkd and everything except the Tuber-only extensions works unchanged. The gem is a fork of beaneater, so it is also a drop-in beaneater replacement — see Migrating from beaneater.

@tuber = Tuber.new('localhost:11300')

@tube = @tuber.tubes["my-tube"]
@tube.put '{"key": "foo"}', pri: 5
@tube.put '{"key": "bar"}', delay: 3

while @tube.peek(:ready)
  job = @tube.reserve
  puts job.body
  job.delete
end

@tuber.close

Install the tuber server (a single binary; brew install tuberq/tuber/tuber, Docker image ghcr.io/tuberq/tuber, or a release binary) — or use an existing beanstalkd — then add the gem:

# Gemfile
gem 'tuber'

Tuber forked from beaneater 1.1.4. The rename is the only breaking change — every method, option and return value is unchanged. Two things to update:

require 'beaneater'                        # before
require 'tuber'                           # after

@beanstalk = Beaneater.new('localhost:11300')  # before
@beanstalk = Tuber.new('localhost:11300')      # after

That includes the nested constants (Beaneater::Job → Tuber::Job), the error classes (Beaneater::NotConnected → Tuber::NotConnected) and the configuration block (Beaneater.configure → Tuber.configure). In most codebases a case-sensitive Beaneater → Tuber and beaneater → tuber replacement is the whole migration.

The BEANSTALKD_URL environment variable is still honoured (it's a beanstalkd convention, not a beaneater one), though TUBER_URL now takes precedence. Likewise config.beanstalkd_url remains as an alias for config.tuber_url.

To setup advanced options for tuber, you can pass configuration options using:

Tuber.configure do |config|
  # config.default_put_delay   = 0
  # config.default_put_pri     = 65536
  # config.default_put_ttr     = 120
  # config.job_parser          = lambda { |body| body }
  # config.job_serializer      = lambda { |body| body }
  # config.tuber_url           = 'localhost:11300'
  # config.connect_timeout     = nil
  # config.resolv_timeout      = nil
  # config.read_timeout        = nil
  # config.write_timeout       = nil
end

The above options are all defaults, so only include a configuration block if you need to make changes.

connect_timeout and resolv_timeout are passed through to TCPSocket.new on Ruby 3.0 and newer (ignored on Ruby < 3.0 for compatibility). read_timeout and write_timeout apply socket read and write timeouts via setsockopt.

@tuber = Tuber.new('10.0.1.5:11300')

# Or use ENV['TUBER_URL'] (or ENV['BEANSTALKD_URL'])
@tuber = Tuber.new

@tuber.close

Tubes are named work queues. Jobs are put into the used tube and reserved from watched tubes. Each tube has a ready, delayed, and buried queue.

@tube = @tuber.tubes.find("some-tube")

# Watch tubes for reserving jobs
@tuber.tubes.watch!('some-tube')        # watch only these tubes
@tuber.tubes.watch('another-tube')       # append to watch list
@tuber.tubes.ignore('some-tube')         # stop watching

# List tubes
@tuber.tubes.all       # => [<Tube name='foo'>, <Tube name='bar'>]
@tuber.tubes.used      # => <Tube name='bar'>
@tuber.tubes.watched   # => [<Tube name='foo'>]

# Manage tubes
@tube.pause(3)        # pause for 3 seconds
@tube.clear           # delete all jobs
@tube.flush           # delete all jobs, returns count
@tube.flush_buried    # delete only buried jobs (Tuber only), returns count

Each client manages two separate concerns: use/using controls where put places jobs, and watch/watching controls where reserve takes jobs from. These are fully orthogonal.

A job has a body (string) and metadata. The typical lifecycle:

   put            reserve               delete
  -----> [READY] ---------> [RESERVED] --------> *poof*

Jobs are in one of three states:

@tube.put "job-data-here"
@tube.put({foo: 'bar'}.to_json)
@tube.put "job-data-here", pri: 1000, delay: 50, ttr: 200

• pri — integer < 2^32, lower values run first (default: 65536)
• delay — seconds to wait before the job becomes ready (default: 0)
• ttr — time to run, seconds a worker has to finish the job (default: 120)

job = @tuber.tubes.reserve      # blocks until a job is available
job = @tuber.tubes.reserve(5)   # wait up to 5 seconds

puts job.body
puts job.tube
puts job.stats.state  # => 'reserved'

job.touch    # extend ttr
job.delete   # success
job.release delay: 5  # retry later
job.bury     # set aside for inspection

@tuber.jobs.find(123)       # peek at a specific job
@tube.peek(:ready)              # peek at next ready job
@tube.peek(:buried)
@tube.peek(:delayed)

@tuber.tubes['some-tube'].kick(3)  # kick 3 buried jobs back to ready

Register handlers for tubes and let process! loop over incoming jobs:

@tuber.jobs.register('some-tube', retry_on: [SomeError]) do |job|
  do_something(job)
end

@tuber.jobs.register('other-tube') do |job|
  do_something_else(job)
end

@tuber.jobs.process!

The loop reserves a job, calls the matching handler, then: deletes on success, releases on retry_on errors, and buries on other exceptions. Raise AbortProcessingError to stop the loop.

When using Tuber, you can group related jobs and chain dependent work using group: and after: options on put. After-jobs are held until every job in the group they depend on has been deleted:

# Fan-out: enqueue grouped work
@tube.put "import-row-1", group: "import"
@tube.put "import-row-2", group: "import"

# Fan-in: this job waits until all "import" jobs are deleted
@tube.put "send-summary", after: "import"

Chain stages together by combining after: and group: on the same job to build a simple DAG pipeline:

@tube.put "row-1", group: "extract"
@tube.put "row-2", group: "extract"
@tube.put "transform", after: "extract", group: "transform"
@tube.put "load",      after: "transform"

Here transform waits for the extract group to finish, then becomes part of the transform group. load waits for transform to complete.

Buried jobs block group completion — kick them to let the group finish. Group names are global and can span multiple tubes.

Prevent duplicate jobs with the idempotency: option. If a job with the same key already exists in the tube, the original job is returned instead of creating a duplicate:

@tube.put "send-report", idempotency: "daily-report"
# => <Tuber::Job id=1 body="send-report">

@tube.put "send-report", idempotency: "daily-report"
# => <Tuber::Job id=1 body="send-report">  (same job, no duplicate created)

The key is scoped to the tube and cleared when the job is deleted, so the same key can be reused afterwards.

Add a cooldown TTL to keep deduplicating for N seconds after deletion — useful for preventing rapid resubmission:

@tube.put "send-report", idempotency: "daily-report", idempotency_ttl: 300

Limit parallel processing of related jobs. When a job with a concurrency key is reserved, other ready jobs sharing the same key are hidden from reserve until the reservation ends:

# Only one job per user can be processed at a time
@tube.put "process-user-42", concurrency: "user-42"
@tube.put "process-user-42-again", concurrency: "user-42"

The second job won't be reserved until the first is deleted, released, or buried. Set a higher limit to allow N concurrent reservations:

# Allow up to 3 concurrent API jobs
@tube.put "api-call-1", concurrency: "api", concurrency_limit: 3
@tube.put "api-call-2", concurrency: "api", concurrency_limit: 3

By default, reserve picks the highest-priority job across all watched tubes. Switch to weighted mode to select tubes randomly in proportion to their weight:

@tuber.tubes.watch('email')
@tuber.tubes.watch('notifications', weight: 2)
@tuber.tubes.watch('batch-jobs', weight: 6)

@tuber.tubes.reserve_mode(:weighted)

job = @tuber.tubes.reserve  # batch-jobs selected 6x as often as email

Tubes default to weight 1. Switch back with reserve_mode(:fifo).

Reserve multiple jobs atomically in a single call:

jobs = @tuber.tubes.reserve_batch(10)  # up to 10 jobs

jobs.each do |job|
  process(job)
  job.delete
end

By default reserve_batch is non-blocking — it returns whatever is ready immediately, possibly an empty array. Pass a timeout (in seconds) to long-poll instead: the call blocks until the first job arrives, then drains everything ready up to count, or returns an empty array when the timeout elapses. This avoids hot-looping a worker on empty polls.

jobs = @tuber.tubes.reserve_batch(10, 30)  # block up to 30s for the first job

While blocked, a positive-timeout batch reserve may raise Tuber::DeadlineSoonError if one of the connection's already-reserved jobs is about to hit its TTR — service that job, then reserve again.

A batch reserve starts the TTR clock on every job at the same instant, but a worker processes them serially — so the tail of a large batch can expire and return to the queue while the worker is still busy. touch_all extends the TTR of every job the connection currently holds in a single command:

jobs = @tuber.tubes.reserve_batch(10)

jobs.each do |job|
  process(job)
  job.delete
  @tuber.jobs.touch_all   # heartbeat whatever is still held
end

No ids are sent: the server tracks the reserved set per connection, so jobs already deleted, released, buried or lost to a TTR timeout are simply absent. Each job keeps its own TTR — deadlines are extended individually, not levelled onto a common value.

The return value is how many jobs the connection actually still holds. A count lower than expected means jobs hit their TTR and went back to the queue while the worker was busy — otherwise invisible, since nothing notifies a worker that it lost a job.

@tuber.stats                            # server-wide stats
@tuber.tubes['some-tube'].stats         # tube stats
@tuber.jobs[some_job_id].stats          # job stats

Tuber.configure do |config|
  config.default_put_delay   = 0
  config.default_put_pri     = 65536
  config.default_put_ttr     = 120
  config.job_parser          = lambda { |body| body }
  config.job_serializer      = lambda { |body| body }
  config.tuber_url           = 'localhost:11300'
end

The job_serializer is applied to every put body — useful for automatic JSON encoding:

Tuber.configure do |config|
  config.job_serializer = lambda { |body| JSON.dump(body) }
end

See the Tuber protocol (a superset of the beanstalk protocol) for additional error types.

• Tuber
• Tuber protocol
• Tuber on RubyGems
• Beanstalkd and the beanstalk protocol
• Backburner — Ruby job queue for Rails/Sinatra

Tuber is maintained by Dan Milne, and builds on the work of everyone who wrote beaneater, from which it is forked:

• Nico Taing - Creator and co-maintainer of beaneater
• Nathan Esquenazi - Contributor and co-maintainer
• Keith Rarick - Much code inspired and adapted from beanstalk-client
• Vidar Hokstad - Replaced telnet with correct TCP socket handling
• Andreas Loupasakis - Improve test coverage, improve job configuration