Mel is an asychronous event-driven jobs processing engine designed to scale. Mel simplifies jobs management by abstracting away the nuances of scheduling and running jobs.
In Mel, a scheduled job is called a task. A single job may be scheduled in multiple ways, yielding multiple tasks from the same job.
Mel schedules all tasks in the chosen storage backend as a set of task ids sorted by their times of next run. For recurring tasks, the next run is scheduled right after the current run completes.
This makes the storage backend the source of truth for schedules, allowing to easily scale out Mel to multiple instances (called workers), or replace or stop workers without losing schedules.
Mel supports bulk scheduling of jobs as a single atomic unit. There's also support for sequential scheduling to track a series of jobs and perform some action after they are all complete.
-
Instant Tasks: These are tasks that run only once after they are scheduled, either immediately or at some specified time in the future.
-
Periodic Tasks: These are tasks that run regularly at a specified interval. They may run forever, or till some specified time in the future.
-
Cron Tasks: These are tasks that run according to a specified schedule in Unix Cron format. They may run forever, or till some specified time in the future.
-
Add the dependency to your
shard.yml:dependencies: mel: github: GrottoPress/mel #redis: # Uncomment if using the Redis backend # github: jgaskins/redis
-
Run
shards update -
Require and configure Mel in your app (we'll configure workers later):
# ->>> src/app/config.cr # ... require "mel" require "../jobs/**" Mel.configure do |settings| settings.error_handler = ->(error : Exception) { puts error.message } settings.timezone = Time::Location.load("Africa/Accra") end Log.setup(Mel.log.source, :info, Log::IOBackend.new) # Redis::Connection::LOG.level = :info # Uncomment if using the Redis backend # ...
-
Using the Redis backend
# ->>> src/app/config.cr # ... require "mel/redis" Mel.configure do |settings| # ... settings.store = Mel::Redis.new( "redis://localhost:6379/0", namespace: "mel" ) # ... end # ...
-
Using the Memory backend (Not for production use)
# ->>> src/app/config.cr # ... require "mel" Mel.configure do |settings| # ... settings.store = Mel::Memory.new # ... end # ...
-
Skip storage
You may disable storage altogether by setting
Mel.settings.storetonil(This is the default).
-
-
Define job:
# ->>> src/jobs/do_some_work.cr struct DoSomeWork include Mel::Job # <= Required def initialize(@arg_1 : Int32, @arg_2 : String) end # <= Instance vars must be JSON-serializable # (Required) # # Main operation to be performed. # Called in a new fiber. def run # << Do work here >> end # Called in the main fiber, before spawning the fiber # that calls the `#run` method above. def before_run # ... end # Called in the same fiber that calls `#run`. # `success` is `true` only if the run succeeded. def after_run(success) if success # ... else # ... end end # Called in the main fiber before enqueueing the task in # the store. def before_enqueue # ... end # Called in the main fiber after enqueueing the task in # the store. `success` is `true` only if the enqueue succeeded. def after_enqueue(success) if success # ... else # ... end end end
-
Schedule job:
-
Run job now:
# ->>> src/app/some_file.cr DoSomeWork.run(arg_1: 5, arg_2: "value") # <= Alias: DoSomeWork.run_now(...)
-
Run job after given delay:
# ->>> src/app/some_file.cr DoSomeWork.run_in(5.minutes, arg_1: 5, arg_2: "value")
The given
Time::Spancan be negative. Eg:DoSomeWork.run_in(-5.minutes, ...). This may be useful for prioritizing certain tasks. -
Run job at specific time:
# ->>> src/app/some_file.cr DoSomeWork.run_at(10.minutes.from_now, arg_1: 5, arg_2: "value")
The specified
Timecan be in the past. Eg:DoSomeWork.run_at(-10.minutes.from_now, ...). This may be useful for prioritizing certain tasks. -
Run periodically:
# ->>> src/app/some_file.cr DoSomeWork.run_every(10.minutes, for: 1.hour, arg_1: 5, arg_2: "value")
This will do the first run 10 minutes from now. If you would like to do the first run some other time, specify that in a
from:argument:# ->>> src/app/some_file.cr DoSomeWork.run_every(10.minutes, from: Time.local, for: 1.hour, arg_1: 5, arg_2: "value")
Instead of
for:, you may usetill:and specify aTime. Leave those out to run forever. -
Run on a Cron schedule:
# ->>> src/app/some_file.cr DoSomeWork.run_on("0 */2 * * *", for: 6.hours, arg_1: 5, arg_2: "value")
This will do the first run relative to now. For instance, if the time now is 03:00, the first run would be at 04:00, the next run at 06:00, and so on. If you would like to do the first run relative to some other time, specify that in a
from:argument:# ->>> src/app/some_file.cr DoSomeWork.run_on("0 */2 * * *", from: 3.days.from_now, for: 6.hours, arg_1: 5, arg_2: "value")
Instead of
for:, you may usetill:and specify aTime. Leave those out to run forever.
The
DoSomeWork.run_*methods accept the following additional arguments:retries: Number of times to attempt a task after it fails, before giving up. This could be specified as a simple integer (eg:3), or a list of backoffs (eg:{2, 4, 1}, or{2.seconds, 4.seconds, 1.second}). Default:{2, 4, 8, 16}. A task fails when any exception is raised during run.
-
-
Start Mel:
-
As its own process (compiled separately):
# ->>> src/worker.cr require "mel" require "./app/**" Mel.configure do |settings| settings.batch_size = -100 settings.poll_interval = 3.seconds settings.worker_id = ENV["WORKER_ID"].to_i end Mel.start # <= Blocks forever, polls for due tasks and runs them. # <= You may stop Mel by sending `Signal::INT` or `Signal::TERM`. # <= Mel will wait for all running tasks to complete before exiting.
-
As part of your app (useful for testing):
# ->>> spec/spec_helper.cr # ... require "mel/spec" Mel.configure do |settings| settings.batch_size = -1 settings.poll_interval = 1.millisecond settings.worker_id = 1 end Spec.before_each { Mel::Task::Query.truncate } Spec.after_suite do Mel.stop Mel::Task::Query.truncate end # <= `Mel.stop` waits for all running tasks to complete before exiting Mel.start_async # ...
-
-
Configure compile targets:
# ->>> shard.yml # ... targets: app: main: src/app.cr worker: main: src/worker.cr # ...
A job's .run_* methods allow scheduling that single job in multiple ways. However, there may be situations where you need to schedule a job the same way, every time.
Mel comes with Mel::Job::Now, Mel::Job::In, Mel::Job::At, Mel::Job::Every and Mel::Job::On templates to do exactly this:
# Define job
struct DoSomeWorkNow
include Mel::Job::Now # <= Required
def initialize(@arg_1 : Int32, @arg_2 : String)
end
# (Required)
def run
# << Do work here >>
end
end
# Schedule job
DoSomeWorkNow.run(arg_1: 5, arg_2: "value")
# <= Alias: `DoSomeWorkNow.run_now(...)`# Define job
struct DoSomeWorkIn
include Mel::Job::In # <= Required
def initialize(@arg_1 : Int32, @arg_2 : String)
end
# (Required)
def run
# << Do work here >>
end
end
# Schedule job
DoSomeWorkIn.run_in(10.minutes, arg_1: 5, arg_2: "value")# Define job
struct DoSomeWorkAt
include Mel::Job::At # <= Required
def initialize(@arg_1 : Int32, @arg_2 : String)
end
# (Required)
def run
# << Do work here >>
end
end
# Schedule job
DoSomeWorkAt.run_at(Time.local(2021, 6, 9, 5), arg_1: 5, arg_2: "value")# Define job
struct DoSomeWorkEvery
include Mel::Job::Every # <= Required
def initialize(@arg_1 : Int32, @arg_2 : String)
end
# (Required)
def run
# << Do work here >>
end
end
# Schedule job
DoSomeWorkEvery.run_every(2.hours, arg_1: 5, arg_2: "value")
# <= Overload: `.run_every 2.hours, for: 5.hours`
# <= Overload: `.run_every 2.hours, till: 9.hours.from_now`# Define job
struct DoSomeWorkOn
include Mel::Job::On # <= Required
def initialize(@arg_1 : Int32, @arg_2 : String)
end
# (Required)
def run
# << Do work here >>
end
end
# Schedule job
DoSomeWorkOn.run_on("0 8 1 * *", arg_1: 5, arg_2: "value")
# <= Overload: `.run_on "0 8 1 * *", for: 100.weeks`
# <= Overload: `.run_on "0 8 1 * *", till: Time.local(2099, 12, 31)`A template excludes all methods not relevant to that template. For instance, calling .run_every or .run_now for a Mel::Job::At template won't compile.
All other methods and callbacks usable in a regular job may be used in a template, including before_* and after_* callbacks.
You may include more than one template in a single job. For instance, including Mel::Job::At and Mel::Job::Every in a job means you can call .run_at and .run_every methods for that job.
Additionally, Mel comes with two grouped templates: Mel::Job::Instant and Mel::Job::Recurring.
Mel::Job::Instant is equivalent to Mel::Job::Now, Mel::Job::In and Mel::Job::At combined. Mel::Job::Recurring is the equivalent of Mel::Job::Every and Mel::Job::On combined.
Mel::Job is itself a grouped template that combines all the other templates.
You may specify an ID whenever you schedule a new job, thus: DoSomeWork.run_*(... id: "1001", ...). If not specified, Mel automatically generates a unique dynamic ID for the task.
Dynamic task IDs may be OK for triggered jobs (jobs triggered by some kind of user interaction), such as a job that sends an email notification whenever a user logs in.
However, there may be jobs that are scheduled unconditionally when your app starts (global jobs). For example, sending invoices at the beginning of every month. You should specify unique static IDs for such tasks.
Otherwise, every time the app (re)starts, jobs are scheduled again, each time with a different set of IDs. The store would accept the new schedules because the IDs are different, resulting in duplicated scheduling of the same jobs.
This is particularly important if you run multiple instances of your app. Hardcoding IDs for global jobs means that all instances hold the same IDs, so cannot reschedule a job that has already been scheduled by another instance.
A task ID may be a mixture of static and dynamic parts. For instance, you may include the current month and year for a global job that runs once a month, to ensure it is never scheduled twice within the same month.
A common pattern is to break up long-running tasks into smaller tasks. For example:
struct SendAllEmails
include Mel::Job
def initialize(@users : Array(User))
end
def run
@users.each { |user| send_email(user) }
end
private def send_email(user)
# Send email
end
end
# Schedule job
users = # ...
SendAllEmails.run(users: users)The above job would run in a single fiber, managed by whichever worker pulls this task at run time. This could mean too much work for a single worker if the number of users is sufficiently large.
Moreover, some mails may be sent multiple times if the task is retried as a result of failure. Ideally, jobs should be idempotent, and as atomic as possible.
The preferred approach is to define a job that sends email to one user, and schedule that job for as many users as needed:
struct SendAllEmails
include Mel::Job
def initialize(@users : Array(User))
end
def run
return if @users.empty?
# Pushes all jobs atomically, at the end of the block.
#
transaction do |store|
# Pass `store` to `.run_*`.
@users.each { |user| SendEmail.run(store: store, user: user) }
end
end
struct SendEmail
include Mel::Job
def initialize(@user : User)
end
def run
send_email(@user)
end
private def send_email(user)
# Send email
end
end
end
# Schedule job
users = # ...
SendAllEmails.run(users: users)
# <= Any `.run_*` method could be called here, as with any job.Bulk scheduling works OK as a fire-and-forget mechanism. However, you may need to keep track of a series of jobs as a single unit, and perform some action only after the last job is done.
This is where sequential scheduling comes in handy. Mel's event-driven design allows chaining jobs, by scheduling the next after the current one completes:
struct SendAllEmails
include Mel::Job
def initialize(@users : Array(User))
end
def run
@users[0]?.try do |user|
send_email(user) # <= Send first email
end
end
def after_run(success)
return unless success
if @users[1]?
self.class.run(users: @users[1..]) # <= Schedule next email
else # <= All emails have been sent
# Do something
end
end
private def send_email(user)
# Send email
end
end
# Schedule job
users = # ...
SendAllEmails.run(users: users)Although the example above involves a single job, sequential scheduling can be applied to multiple different jobs, each representing a step in a workflow, with each job scheduling the next job in its #after_run callback:
struct SomeJob
include Mel::Job
def run
# Do something
end
def after_run(success)
SomeStep.run if success
end
struct SomeStep
include Mel::Job
def run
# Do something
end
def after_run(success)
SomeOtherStep.run if success
end
end
struct SomeOtherStep
include Mel::Job
def run
# Do something
end
def after_run(success)
# All done; do something
end
end
endMel provides a progress tracker for jobs. This is particularly useful for tracking multiple jobs representing a series of steps in a workflow:
# ->>> src/app/config.cr
# ...
Mel.configure do |settings|
settings.progress_expiry = 1.day
end
# ...# ->>> src/jobs/some_job.cr
struct SomeJob
include Mel::Job
def initialize
@progress = Mel::Progress.start(id: "some_job", description: "Awesome job")
end
# ...
def after_run(success)
return @progress.fail unless success
transaction do |store|
SomeStep.run(store: store, progress: @progress)
@progress.move(50, store) # <= Move to 50%
end
end
struct SomeStep
include Mel::Job::Now
def initialize(@progress : Mel::Progress)
end
# ...
def after_run(success)
return @progress.fail unless success
transaction do |store|
SomeOtherStep.run(store: store, progress: @progress)
@progress.move(80, store) # <= Move to 80%
end
end
end
struct SomeOtherStep
include Mel::Job::Now
def initialize(@progress : Mel::Progress)
end
# ...
def after_run(success)
return @progress.fail unless success
@progress.succeed # <= Move to 100%
end
end
end
# Schedule job
SomeJob.run
# Track progress
#
# This may, for instance, be used in a route in a web application.
# Client-side javascipt can query this route periodically, and
# show response using a progress tracker UI.
#
report = Mel::Progress.track("some_job")
report.try do |_report|
_report.description
_report.id
_report.value
_report.failure?
_report.running?
_report.success?
_report.started?
_report.ended?
endYou may delete progress data in specs thus:
# ->>> spec/spec_helper.cr
# ...
Spec.before_each do
# ...
Mel::Progress::Query.truncate
# ...
end
Spec.after_suite do
# ...
Mel::Progress::Query.truncate
# ...
end
# ...A Mel worker waits for all running tasks to complete before exiting, if it received a Signal::INT or a Signal::TERM, or if you called Mel.stop somewhere in your code. This means jobs are never lost mid-flight.
Jobs are not lost even if there is a force shutdown of the worker process, since Mel does not delete a task from the store until it is complete. The worker can pick off where it left off when it comes back online.
Mel relies on the worker_id setting to achieve this. Each worker, therefore, must set a unique, static integer ID, so it knows which pending tasks it owns.
Once a task enters the pending state, only the worker that put it in that state can run it. So if you need to take down a worker permanently, ensure that it completes all pending tasks by sending the appropriate signal.
Because each worker requires it's own unique .worker_id, autoscaling as used in classic distributed architectures should not be used, since auto-scaled replicas would inherit the same configuration as the original instance.
This would lead to multiple workers using the same .worker_id, which could result in pending jobs being run multiple times; once each for each replica that starts up.
Instead, it is recommended that a new service be registered for each worker that is to be deployed, and the appropriate .worker_id set for each.
-
Using
Procfile:# ->> Procfile # ... worker_1: export WORKER_ID=1 && ./bin/worker worker_2: export WORKER_ID=2 && ./bin/worker worker_3: export WORKER_ID=3 && ./bin/worker # ...
-
Using docker compose for swarm:
# ->> docker-compose.yml # ... services: worker_1: command: ./bin/worker environment: WORKER_ID: "1" deploy: replicas: 1 worker_2: command: ./bin/worker environment: WORKER_ID: "2" deploy: replicas: 1 worker_3: command: ./bin/worker environment: WORKER_ID: "3" deploy: replicas: 1 # ...
Another option is to accept the worker ID as a command argument:
# ->> src/worker.cr
# ...
ARGV.first?.try { |worker_id| Mel.settings.worker_id = worker_id.to_i }
Mel.start-
Using
Procfile:# ->> Procfile # ... worker_1: ./bin/worker 1 worker_2: ./bin/worker 2 worker_3: ./bin/worker 3 # ...
-
Using docker compose for swarm:
# ->> docker-compose.yml # ... services: worker_1: command: ./bin/worker 1 deploy: replicas: 1 worker_2: command: ./bin/worker 2 deploy: replicas: 1 worker_3: command: ./bin/worker 3 deploy: replicas: 1 # ...
-
Using config for Fly.io:
# ->> fly.toml # ... [processes] worker_1 = './bin/worker 1' worker_2 = './bin/worker 2' worker_3 = './bin/worker 3' # ...
Ensure no spare machines are created by passing
--ha=falsetofly deploycommand.
Mel's batch_size setting allow setting a limit on the number of due tasks to retrieve and run each poll, and, consequently, the number of fibers spawned to handle those tasks.
If the setting is a positive integer N, Mel would pull and run N due tasks each poll.
If it is a negative integer -N (other than -1), the number of due tasks pulled and ran each poll would vary such that the total number of running tasks would not be greater than N.
-1 sets no limits. Mel would pull as many tasks as are due each poll, and run all of them.
Link: https://github.com/luckyframework/carbon
-
Require
mel/carbon, after your emails:# ->>> src/app.cr # ... require "emails/base_email" require "emails/**" require "mel/carbon" # ...
-
Set up base email:
# ->>> src/emails/base_email.cr abstract class BaseEmail < Carbon::Email # ... include JSON::Serializable # ... end
-
Configure deliver later strategy:
# ->>> config/email.cr BaseEmail.configure do |settings| # ... settings.deliver_later_strategy = Mel::Carbon::DeliverLater.new # ... end
Create a .env.sh file:
#!/bin/bash
export REDIS_URL='redis://localhost:6379/0'Update the file with your own details. Then run tests with source .env.sh && crystal spec -Dpreview_mt.
- Fork it
- Switch to the
masterbranch:git checkout master - Create your feature branch:
git checkout -b my-new-feature - Make your changes, updating changelog and documentation as appropriate.
- Commit your changes:
git commit - Push to the branch:
git push origin my-new-feature - Submit a new Pull Request against the
GrottoPress:masterbranch.