crumble-jobs

Background job framework for Crumble applications.

Installation

1. Add the dependency to your shard.yml:

   dependencies:
     crumble-jobs:
       github: sbsoftware/crumble-jobs
   

2. Run shards install.

Usage

require "crumble-jobs"

Configure the queue

Configure a single queue backend for the app using the macro. The queue is reused by jobs and workers.

Crumble::Jobs.configure_queue Crumble::Jobs::InMemoryQueue.new
# or
Crumble::Jobs.configure_queue Crumble::Jobs::FileQueue.new("./tmp/jobs")

After configuration, jobs enqueue through Crumble::Jobs.queue and workers default to the same queue.

Define jobs

class IncrementCounterJob < Crumble::Jobs::Job
  params session_key : String, counter_id : Int64

  def perform : Nil
    # ...
  end
end

Use params to declare jobs with arguments. Jobs without arguments do not need a params declaration.

Throttle a job class:

class SyncWebhookJob < Crumble::Jobs::Job
  throttle max_jobs: 10, timespan: 5.minutes
  params endpoint : String

  def perform : Nil
    # ...
  end
end

throttle applies when a worker starts jobs, not when jobs are enqueued. Enqueueing is never blocked by throttling. A worker starts at most max_jobs jobs of that class inside one window, and the window starts with the first execution in the current burst. After the limit is reached, later jobs of the same class wait until the window expires, and same-class execution order is preserved.

Enqueue a job:

IncrementCounterJob.enqueue("session-uuid", 123)
IncrementCounterJob.enqueue_in(5.minutes, "session-uuid", 123)
IncrementCounterJob.enqueue_at(Time.utc + 1.hour, "session-uuid", 123)

Declare a zero-argument job as periodic to seed it when the worker starts and schedule its next occurrence after each run:

class RefreshPricesJob < Crumble::Jobs::Job
  periodic every: 5.minutes

  def perform : Nil
    # ...
  end
end

Periodic jobs are paced from the completion of one occurrence to the start of the next. Failed occurrences also schedule the next run. FileQueue checks ready and currently processing jobs when seeding, preventing a restart from creating a parallel chain. Applications that manage their own worker loop can call Crumble::Jobs.seed_periodic_jobs explicitly.

Worker

worker = Crumble::Jobs::Worker.new(max_concurrency: 4, shutdown_timeout: 30.seconds)
worker.start

start handles SIGTERM and SIGINT by stopping new reservations and waiting up to shutdown_timeout for active jobs. Reservations still running after the timeout are returned to the queue. FileQueue also recovers reservations abandoned by crashes or forced termination while preserving reservations locked by other live worker processes.

Development

• Format: crystal tool format
• Tests: crystal spec

Contributing

1. Fork it (https://github.com/sbsoftware/crumble-jobs/fork)
2. Create your feature branch (git checkout -b my-new-feature)
3. Commit your changes (git commit -am 'Add some feature')
4. Push to the branch (git push origin my-new-feature)
5. Create a new Pull Request

Contributors

• Stefan Bilharz - creator and maintainer