🚀 Cosmonats

Background jobs + real-time event streaming for Ruby — unified, in one gem, backed by NATS.
No Redis. No DB polling. Disk-backed, horizontally scalable — no message is ever silently dropped.

Battle-tested in production. Tens of millions of jobs processed and counting.

⚡ Taste it

# Define a job with a familiar look
class SendEmailJob
  include Cosmo::Job
  options stream: :default, retry: 3, dead: true

  def perform(user_id, template)
    EmailService.send(user_id, template)
  end
end

# Enqueue it
SendEmailJob.perform_async(123, "welcome")
SendEmailJob.perform_in(1.day, 123, "followup")

# Process a continuous real-time event stream
class ClicksProcessor
  include Cosmo::Stream
  options stream: :clickstream, batch_size: 100,
          consumer: { subjects: ["events.clicks.>"] }

  def process_one
    Analytics.track(message.data)
    message.ack
  end
end

ClicksProcessor.publish({ user_id: 123, page: "/home" }, subject: "events.clicks.homepage")

bundle exec cosmo -C config/cosmo.yml -c 20         # Run jobs + streams with 20 threads
bundle exec cosmo -C config/cosmo.yml -c 20 jobs    # Jobs only
bundle exec cosmo -C config/cosmo.yml -c 20 streams # Streams only

📖 Index

Why?
Features
Installation
Quick Start
Core Concepts
Advanced Usage
CLI Reference
Deployment
Monitoring
Examples

🎯 Why?

Most background job libraries use Redis or Postgres — tools that were never designed for this. Think of NATS as Redis — but Redis is KV first then messaging; NATS is messaging first, then KV. What NATS is:

~20 MB binary, ~10 MB RAM at idle Trivial to run anywhere.
Disk-backed persistent streams Messages survive restarts, don't require RAM to fit.
True horizontal clustering Lose a node — other nodes take over, zero message loss.
Multilingual Official clients for Ruby, Go, Python, Rust, Java, .NET, and more. Any service can publish or consume.

One NATS server replaces your message broker, job queue, and KV store — with lower operational overhead.

	Redis/DB-backed	NATS/Cosmonats
Persistence	In-memory / DB bloat	Disk-backed, TB-scale
Scaling	Sentinel only / Vertical only	True horizontal clustering
Background jobs	Yes	Yes
Real-time stream	No	Yes
Zero message loss	No	Yes
Message replay	No	Yes
Backpressure	No, grow unbounded	Yes
Multi-DC	Complex setup	Native geo-distribution

Killer Features:

— Jobs + Streams, unified in one gem.

Most Ruby gems handle exactly that — background jobs. If you also need to consume a continuous event feed, that's a second system, second config, second set of worker processes, second Dockerfile entry. Cosmonats is the only Ruby gem with a first-class Job primitive and a first-class Stream primitive, sharing one server, one config, one CLI, one monitoring endpoint.

— Message replay and time-travel debugging.

NATS persists messages to disk and lets any consumer rewind to any point — beginning of time, a specific timestamp, or only new messages.

Incident recovery — your pipeline crashed for 3 hours. Replay from the crash timestamp.
New consumer bootstrap — a new service needs historical events. Start it from the beginning.
Bug reproduction — replay the exact sequence of messages that caused a production issue.

— Multi-datacenter queues, natively.

NATS has a first-class cluster + leaf-node architecture for geo-distribution. Spanning multiple regions or datacenters is a config block — not a separate product or a third-party replication tool. NATS was built for edge computing, IoT, and satellite communication — multi-DC is a first-class concern, not an afterthought.

— Transport-level deduplication + built-in KV. No extra infrastructure.

NATS deduplicates messages at the broker — same-ID messages within the configured window are dropped before they ever reach a worker. No uniqueness gems, no advisory locks, no extra round-trips. It also ships a built-in Key/Value store usable for distributed locks and rate limiting — no Redis, no Memcached, nothing else to run.

✨ Features

🎪 Job Processing

Familiar API — perform_async, perform_in, perform_at
Priority queues — critical, high, default, low with weighted round-robin
Scheduled jobs — execute at a specific time or after a delay
Automatic retries — exponential backoff, configurable attempts
Dead letter queue — capture permanently failed jobs
Job uniqueness — prevent duplicate execution
Concurrency limits — cap simultaneous executions per class or per key
Cron scheduling — recurring jobs manageable live from the web UI
Batches — group jobs and fire a callback once the whole group finishes, including nested batches

🌊 Stream Processing

Real-time event streams — process continuous data feeds
Batch processing — handle multiple messages in one go
Message replay — reprocess from any point in time
Consumer groups — load-balanced across workers
Custom serialization — JSON, MessagePack, Protobuf
Pause / resume — stop and restart a stream's processing without losing its position

📦 Installation

# Gemfile
gem "cosmonats"

Requirements: Ruby ≥ 3.1, NATS Server ≥ 2.14 with JetStream enabled (install guide)

Optional gems — not installed with cosmonats, add them only for the features you use:

Feature	Gems
Monitoring UI (`Cosmo::Web`)	`rack`
Health check HTTP server (`/health`, `/ping`)	`rack`, `webrick`

# Gemfile
gem "rack", "~> 3.0"
gem "webrick", "~> 1.9"

Spin up NATS instantly with Docker — one command, that's it:

docker run -p 4222:4222 -p 8222:8222 nats:alpine -js

Or add it to your existing docker-compose.yml:

services:
  nats:
    image: nats:alpine
    command: -js
    ports:
      - "4222:4222"
      - "8222:8222"

Mount the monitoring UI in your Rack app:

require "cosmo/web"

# Rails
mount Cosmo::Web => "/cosmo"

# Any Rack app (config.ru)
map "/cosmo" { run Cosmo::Web }

Warning

Cosmo::Web ships with no authentication and no CSRF protection, yet it exposes destructive actions (retrying and deleting jobs, pausing streams, running and deleting crons). Never expose it unprotected.

# Rails, with Devise
authenticate :user, ->(user) { user.admin? } do
  mount Cosmo::Web => "/cosmo"
end

# Any Rack app (config.ru), with HTTP Basic auth
map "/cosmo" do
  use Rack::Auth::Basic, "Cosmo" do |user, password|
    Rack::Utils.secure_compare(user, ENV.fetch("COSMO_WEB_USER")) &
      Rack::Utils.secure_compare(password, ENV.fetch("COSMO_WEB_PASSWORD"))
  end
  run Cosmo::Web
end

🚀 Quick Start

1. Configure (optional)

Nothing is required: without config/cosmo.yml, Cosmo runs one default job stream plus its own service streams. To change anything, write Cosmo's defaults into your project and edit them:

bundle exec cosmo --init # writes config/cosmo.yml

2. Create streams in NATS (one-time, idempotent)

bundle exec cosmo -S

3. Define a job in `app/jobs/`

class SendEmailJob
  include Cosmo::Job
  options stream: :default, retry: 3, dead: true

  def perform(user_id, email_type)
    UserMailer.send(email_type, user_id).deliver_now
  end
end

4. Enqueue & run

SendEmailJob.perform_async(42, "welcome")

bundle exec cosmo -C config/cosmo.yml -c 10 -r ./app/jobs jobs

💡 Core Concepts

Jobs

class ReportJob
  include Cosmo::Job

  options(
    stream: :critical,  # Stream name
    retry: 5,           # Retry attempts
    dead: true          # Send to dead letter queue on final failure
  )

  def perform(report_id)
    logger.info "Processing report #{report_id}"
    Report.find(report_id).generate!
  rescue StandardError => e
    logger.error "Failed: #{e.message}"
    raise  # Triggers retry with exponential backoff
  end
end

ReportJob.perform_async(42)                              # Enqueue now
ReportJob.perform_in(30.minutes, 42)                     # Delayed
ReportJob.perform_at(Time.parse("2026-01-25 10:00"), 42) # Scheduled
ReportJob.perform_sync(42)                               # Inline, no NATS (great for tests)

Inside perform, the job instance knows about the message it came from:

Accessor	Value
`jid`	Job id, returned by `perform_async`
`attempt`	1-based delivery attempt
`enqueued_at`	`Time` the job was enqueued; for a cron job, when NATS fired the schedule
`scheduled_by`	Subject of the cron schedule that fired the job, `nil` for jobs enqueued any other way
`batch_id`	Id of the batch the job belongs to, `nil` outside a batch

Streams

class ClicksProcessor
  include Cosmo::Stream

  options(
    stream: :clickstream,
    batch_size: 100,
    start_position: :last,  # :first, :last, :new, or timestamp
    consumer: {
      ack_policy: "explicit",
      max_deliver: 3,
      max_ack_pending: 100,
      subjects: ["events.clicks.>"]
    }
  )

  # Process one message at a time
  def process_one
    Analytics.track_click(message.data)
    message.ack
  end

  # OR process a batch
  def process(messages)
    Analytics.bulk_track(messages.map(&:data))
    messages.each(&:ack)
  end
end

# Publishing
ClicksProcessor.publish({ user_id: 123, page: "/home" }, subject: "events.clicks.homepage")

# Acknowledgment strategies
message.ack                          # Success
message.nack(delay: 5_000_000_000)   # Retry in 5 seconds (nanoseconds)
message.term                         # Permanent failure, no retry

Configuration

NATS subjects follow a dot-separated hierarchy (events.clicks.homepage). The > wildcard matches everything after that prefix. Think of subjects as topic names — flexible routing with no extra configuration.

config/cosmo.yml is optional. Cosmo ships its defaults in lib/cosmo/config/cosmo.yml (cosmo --init copies it into your project), and your file is deep-merged over them: keep only what you change. The one list that is not merged is your job streams — listing any under setup.jobs replaces the built-in default stream and its consumer:

concurrency: 10

stream_config: &stream_config
  storage: file
  retention: workqueue
  duplicate_window: 120 # seconds
  discard: new
  allow_direct: true    # required by the web UI
  subjects: ["jobs.%{name}.>"]

consumer_config: &consumer_config
  ack_policy: explicit
  max_deliver: 30       # keep it above every job class's retry
  max_ack_pending: 20
  ack_wait: 60          # seconds
  subject: jobs.%{name}.>

setup:
  jobs:
    critical: { <<: *stream_config }
    default: { <<: *stream_config }
  streams:              # Cosmo::Stream processors are always yours to declare
    events:
      storage: file
      retention: limits
      max_age: 86400
      subjects: ["events.>"]

consumers:
  jobs:
    critical: { <<: *consumer_config, priority: 50 }
    default: { <<: *consumer_config, priority: 15 }

max_age, duplicate_window, and ack_wait are in seconds. Without setup.jobs, consumers.jobs.default still tunes the built-in consumer (e.g. just ack_wait: 300).

Service streams — the scheduled stream (delayed jobs, crons), the dead stream (dead letters), and the stats counters and KV buckets are Cosmo's own: they are not part of cosmo.yml, and cosmo --setup creates them next to yours. Tune them in Cosmo.configure:

Cosmo.configure do |config|
  config.replicas = 3                # every service stream and bucket, for a NATS cluster (default 1)
  config.dead.max_age = "14d"        # dead-letter retention: max_age, max_msgs (10_000), max_bytes (-1); default 7d
  config.dead.enabled = false        # drop jobs that give up instead of parking them
  config.scheduled.enabled = false   # no scheduled stream or scheduler: perform_in/perform_at and crons raise
  config.batches.expiry = "1d"       # how long batch tracking data lives (default 3d)
  config.metrics.retention = "90d"   # how long the Metrics tab's per-day numbers are kept (default 30d)
  config.metrics.enabled = false     # record nothing and hide the Metrics tab
end

Turning scheduled or dead off also hides their pages in the web UI. Stats can't be turned off — the web UI depends on them.

Programmatic: cosmo.yml holds streams, consumers, and process settings; Cosmo.configure holds what YAML can't express (loggers, middleware) and Cosmo's service settings. Command-line flags (-c, -t, -p) win over the file.

Cosmo.configure do |config|
  config.logger = Rails.logger # any ::Logger; defaults to stdout
  config.log_level = :debug    # COSMO_LOG_LEVEL, when set, wins
end

Cosmo leaves a custom logger's formatter alone, so its lines lack Cosmo's jid/elapsed context unless you set logger.formatter = Cosmo::Logger::SimpleFormatter.new. Trace output is dropped for loggers without a trace level.

Environment variables:

Variable	Default	Description
`NATS_URL`	`nats://localhost:4222`	NATS server(s) to connect to
`COSMO_LOG_LEVEL`	`info`	`trace`, `debug`, `info`, `warn`, `error` or `fatal`
`COSMO_CLIENT_NAME`	`cosmo-<program>-<host>-<pid>`	NATS connection name, shown in `nats server report connections` and `/connz`
`COSMO_JS_TIMEOUT`	`5`	JetStream API timeout in seconds; raise it for the web UI on a busy server
`COSMO_CONNECT_TIMEOUT`	`2`	NATS TCP connect timeout in seconds
`NATS_MAX_RECONNECT_ATTEMPTS`	`-1`	Reconnect attempts before giving up on NATS; `-1` keeps trying through any outage
`COSMO_JOBS_STREAMS`	all configured	Comma-separated job streams to subscribe to, when `--streams`/`--stream` isn't given
`COSMO_JOBS_FETCH_TIMEOUT`	`0.1`	Seconds a jobs worker waits on each fetch
`COSMO_WEB_POLL_INTERVAL`	`5`	Web UI auto-refresh interval in seconds
`COSMO_WEB_CACHE_TTL`	`2`	Seconds the web UI shares a refresh between viewers; `0` turns sharing off

🔧 Advanced Usage

Cron

Recurring jobs, without a separate scheduler process. A schedule is just a message parked in the scheduled stream (requires NATS Server 2.14+) — NATS fires it on the cron expression, and the worker's scheduler dispatches the job on to the stream that runs it, the same path a delayed job takes. Deploy it once; whatever's in NATS is exactly what runs and exactly what shows up in the web UI's Crons tab, where each entry can be inspected, run immediately, or deleted.

Every schedule lives in scheduled because NATS only lets a schedule fire at a subject its own stream covers, and refuses discard: new on any stream with scheduling enabled. Keeping schedules in one stream leaves every stream that actually runs jobs free to choose its own discard policy. The trade is that cron firings are dispatched by the scheduler, so at least one worker has to be running without --no-scheduler.

Declare schedules right in config/cosmo.yml:

setup:
  cron:
    daily_report:
      class: ReportJob
      schedule: "@daily"        # @-shortcuts are passed straight through to NATS
      stream: default
    weekday_digest:
      class: ReportJob
      schedule: "0 9 * * 1-5"   # 6-field NATS cron (seconds first); 5-field UNIX cron is auto-normalized
      stream: default
      args: ["daily"]
      timezone: America/New_York # optional, cron expressions only

cosmo -C config/cosmo.yml -S syncs it — whatever's in the file is exactly what ends up scheduled in NATS, same as streams.

Prefer to manage schedules at runtime instead? The same operations are available from Ruby:

Cosmo::API::Cron.instance.upsert!(
  class_name: "ReportJob", stream: "default", schedule: "0 9 * * 1-5",
  args: ["daily"], timezone: "America/New_York", name: "weekday_report"
)

Cosmo::API::Cron.instance.all                                    # every schedule currently deployed
Cosmo::API::Cron.instance.run_now!("cosmo.cron.default.report_job.weekday_report")  # bypass the timer
Cosmo::API::Cron.instance.delete!("cosmo.cron.default.report_job.weekday_report")   # stop future firings

Priority Queues

class UrgentJob
  include Cosmo::Job
  options stream: :critical  # priority: 50 in config — polled most frequently
end

Batches

Group jobs together and fire a callback once every one of them has finished. The registered class is plain Ruby — it implements on_complete(status, opts) / on_success(status, opts), not perform — and runs on the job worker pool, not inline on whichever thread finalized the batch. status is { bid:, total:, succeeded:, failed: } and opts is whatever you passed to #on.

batch = Cosmo::Batch.new
batch.jobs do
  ImportJob.perform_async(1)
  ImportJob.perform_async(2)
end
batch.on(:complete, NotifyUser, user_id: 1)  # fires once every job has finished, pass or fail
batch.on(:success, NotifyUser, user_id: 1)   # fires only if none of them failed

class NotifyUser
  def on_complete(status, opts)
    UserMailer.batch_done(opts[:user_id], status[:succeeded], status[:total]).deliver_later
  end

  def on_success(status, opts)
    UserMailer.batch_succeeded(opts[:user_id]).deliver_later
  end
end

#jobs is the only place membership is tracked — jobs enqueued outside the block aren't part of the batch. Call it at least once (an empty block is fine) to close the batch.
:complete always fires once every job is done. :success fires only if none were dead-lettered or dropped after exhausting retries. Both can be registered before or after #jobs — a callback registered after the batch has already finished still fires.

Nested batches — a running job can spawn its own sub-batch, which counts as one pending unit of its parent and propagates any failure upward as a single failure unit:

class ImportJob
  include Cosmo::Job

  def perform(account_id)
    sub_batch = Cosmo::Batch.new(parent: batch_id)  # batch_id is this job's own batch, if any
    sub_batch.jobs { SyncRecordJob.perform_async(account_id) }
  end
end

Batch tracking data (pending counts, callbacks, results) expires automatically after config.batches.expiry (default: 3 days), set in Cosmo.configure.
Open and finished batches — with pending/succeeded/failed counts — are listed live on the web UI's Batches page.
Only Cosmo::Job-based jobs are tracked; the ActiveJob adapter doesn't currently participate in batches (see docs/active_job.md).

Concurrency Limiting

class ThirdPartyApiJob
  include Cosmo::Job
  # At most 3 instances of this job run at once, cluster-wide.
  # Jobs that lose the race are NAK'd with a delay equal to `duration`
  # so they aren't redelivered until a slot is guaranteed free.
  options limit: { duration: 30, concurrency: 3 }
end

class PerAccountSyncJob
  include Cosmo::Job
  # Scope the cap per key instead of class-wide — e.g. one concurrent sync per account.
  options limit: { duration: 30, concurrency: { to: 1, key: ->(account_id) { account_id } } }

  def perform(account_id)
    Account.find(account_id).sync!
  end
end

Custom Serializers

module MessagePackSerializer
  def self.serialize(data) = MessagePack.pack(data)
  def self.deserialize(payload) = MessagePack.unpack(payload)
end

class FastStream
  include Cosmo::Stream
  options publisher: { serializer: MessagePackSerializer }
end

Error Handling

class ResilientJob
  include Cosmo::Job
  options retry: 5, dead: true

  def perform(data)
    process_data(data)
  rescue RetryableError => e
    logger.warn "Retryable: #{e.message}"
    raise  # Will retry with exponential backoff
  rescue FatalError => e
    logger.error "Fatal: #{e.message}"
    # Don't raise — won't retry, won't go to DLQ
  end
end

By default, a failed job is redelivered after attempt**4 + 15 seconds. Override that per job class with retry_in, given the 1-based attempt count and the exception that was raised:

class ThrottledApiJob
  include Cosmo::Job
  options retry: 5, retry_in: ->(count, exception) { exception.is_a?(RateLimitedError) ? 60 : count * 10 }

  def perform(...)
    # ...
  end
end

If the proc returns something non-numeric/non-positive, or raises, the default backoff is used instead. Note: if this job class also sets limit: { concurrency: ... } (see above), count includes deliveries that were turned away for lack of a free slot, not just failed attempts.

Error handlers receive every error Cosmo rescues — each failed job attempt, a failing stream batch, fetch and scheduler errors, and rejected messages — so one hook reports them all:

Cosmo.configure do |config|
  config.error_handlers << ->(error, context) { Honeybadger.notify(error, context: context) }
end

context is a Hash with a :source (:job, :stream, :fetch, :scheduler, :reject, :retry_in, :limit) and what is known there: for a job its payload (jid, class, args, …), stream, subject, and attempt; for a stream batch its stream, processor, and size. A handler that raises is logged and skipped.

Middleware

Every job execution on a worker runs through the server middleware chain. A middleware gets the job instance, its parsed payload, and the NATS message, and yields to run the rest of the chain:

class TimingMiddleware
  def initialize(threshold:)
    @threshold = threshold
  end

  def call(job, data, message)
    started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    yield
  ensure
    elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started
    job.logger.warn("#{data[:class]} took #{elapsed.round(2)}s on #{message.metadata.stream}") if elapsed > @threshold
  end
end

Cosmo.configure do |config|
  config.server_middleware do |chain|
    chain.add TimingMiddleware, threshold: 5
  end
end

Register middleware at boot, before workers start (e.g. in a Rails initializer; loading cosmo.yml afterwards keeps it). The chain starts with Cosmo's required middleware: Cosmo::Middleware::Limit (a job class's limit: option), Cosmo::Middleware::Busy (the Web UI's Busy page), Cosmo::Middleware::Totals (processed/failed counts, one per execution), and Cosmo::Middleware::Metrics (the Metrics tab). add appends innermost, after them; prepend, insert_before, insert_after, and remove reorder the chain, so moving or removing the built-ins is at your own risk. A middleware that raises sends the job through the usual retry/DLQ path, and one that doesn't yield skips the job and acks it. To put a job back for later instead, raise Cosmo::Job::Requeue.new(delay_in_seconds): the job is redelivered without failing, though the redelivery still counts toward its retry and max_deliver. A fresh instance is built for every job. perform_sync doesn't run the chain.

The client chain runs where a job is enqueued (perform_async, perform_in, perform_at, and the ActiveJob adapter), around the publish. It starts empty, and a middleware gets the job class name, the payload Hash that will be published, and the stream the job runs on:

class RequestIdMiddleware
  def call(_job_class, payload, _stream)
    payload[:request_id] = Current.request_id
    yield
  end
end

Cosmo.configure do |config|
  config.client_middleware { |chain| chain.add RequestIdMiddleware }
end

Not yielding stops the publish: perform_async returns nil, and a batch the job was joining doesn't wait for it.

Testing

# Synchronous — no NATS needed
SendEmailJob.perform_sync(123, "test")

# Async — returns a job ID
jid = SendEmailJob.perform_async(123, "welcome")
assert_kind_of String, jid

Integrations

ActiveJob:

# config/application.rb
config.active_job.queue_adapter = :cosmonats

The ActiveJob queue name maps directly to a Cosmo stream. Use cosmo_options for anything Cosmo-specific — retries, DLQ behavior, overriding the target stream, or a custom retry delay:

class ReportJob < ApplicationJob
  cosmo_options retry: 5, dead: false, stream: :critical, retry_in: ->(count, exception) { count * 10 }

  def perform(report_id)
    Report.find(report_id).generate!
  end
end

Inside a Rails app this is wired up automatically by the bundled Railtie — it registers the adapter and loads config/cosmo.yml if present. Outside Rails:

require "cosmo/active_job"
ActiveJob::Base.queue_adapter = Cosmo::ActiveJobAdapter::Adapter.new

Sentry:

require "cosmo/middleware/sentry"

Cosmo.configure do |config|
  config.server_middleware { |c| c.add Cosmo::Middleware::Sentry }
  config.error_handlers << Cosmo::Middleware::Sentry::ERROR_HANDLER
end

The middleware wraps every job execution in a Sentry transaction (queue.cosmonats), and the error handler captures every error Cosmo rescues — failed jobs, stream batches, fetches, the scheduler — with its context under cosmonats. No other setup beyond having sentry-ruby initialized.

🖥️ CLI Reference

cosmo -C config/cosmo.yml --setup                     # Create streams in NATS (idempotent)
cosmo -C config/cosmo.yml -c 20 -r ./app/jobs jobs    # Jobs only
cosmo -C config/cosmo.yml -c 20 streams               # Streams only
cosmo -C config/cosmo.yml -c 20                       # Both
cosmo -C config/cosmo.yml jobs --streams default,high # Jobs, limited to these streams
cosmo -C config/cosmo.yml jobs --stream critical      # Jobs, limited to one stream
cosmo -C config/cosmo.yml jobs --no-scheduler         # Jobs, without dispatching scheduled jobs
COSMO_JOBS_STREAMS=default,high cosmo jobs            # Same filter, without touching the command line

Each selected stream gets its own durable consumer, named consumer-<stream>. Every process that subscribes to a stream pulls from that same consumer, so running a dedicated fleet per stream splits the work instead of duplicating it. The scheduled stream is a service stream: it is always dispatched, and naming it in --streams selects nothing, use --no-scheduler on workers that should not dispatch scheduled jobs. Unknown stream names abort the process.

Global flags (before the command):

Flag	Description	Example
`-C, --config PATH`	Config file path	`-C config/cosmo.yml`
`-c, --concurrency INT`	Worker threads	`-c 20`
`-r, --require PATH`	Auto-require directory	`-r ./app/jobs`
`-t, --timeout NUM`	Shutdown timeout (sec)	`-t 60`
`-p, --http-port INT`	Serve `/health` and `/ping` on a port	`-p 9090`
`-S, --setup`	Setup streams & sync cron, then exit	`--setup`
`--init`	Write `config/cosmo.yml` with the defaults, then exit	`--init`
`-v, --version`	Print version and exit	`--version`
`-h, --help`	Show help and exit	`--help`

jobs command options:

Flag	Description	Example
`--streams NAMES`	Only subscribe to these job streams, instead of all configured ones	`--streams default,high`
`--stream NAME`	Same as `--streams`, for a single stream	`--stream default`
`--[no-]scheduler`	Dispatch scheduled jobs (default: yes)	`--no-scheduler`
`--subject NAME`	Job's subject	`--subject jobs.default.foo`

--streams/--stream also read from COSMO_JOBS_STREAMS (comma-separated) when neither flag is given.

streams command options (--stream/--subject/--consumer_name/--batch_size apply only when running a single processor; ignored once --processors selects more than one):

Flag	Description	Example
`--processors NAMES`	Only run these stream processor classes	`--processors OrderProcessor`
`--stream NAME`	Stream name	`--stream orders`
`--subject NAME`	Subject name	`--subject orders.created`
`--consumer_name NAME`	Consumer name	`--consumer_name orders-consumer`
`--batch_size NUM`	Messages per fetch batch	`--batch_size 50`

Signals

Signal	Effect
`INT`, `TERM`	Graceful shutdown: stop fetching, wait up to `--timeout` seconds for in-flight work, exit
`TSTP`	Quiet: stop fetching new work, keep running in-flight work
`CONT`	Resume fetching after `TSTP`
`USR1`	Quiet, then exit once in-flight work drains
`TTIN`	Log every thread's backtrace at `WARN`, tagged with the same `tid` as regular log lines

kill -TSTP <pid>  # before a deploy: let the old worker finish what it has
kill -TTIN <pid>  # a worker looks stuck: see what each thread is doing

The worker's state (running, quiet, stopping) is shown on the web UI's Processes list.

Health check

Pass -p/--http-port (or set http.port in the config) to start a small WEBrick HTTP server alongside the workers. It needs the optional rack and webrick gems (see Installation); without a port they're never loaded.

http:
  port: 9090
  host: 0.0.0.0 # default

GET /health returns 200 when processors are running and NATS is connected, 503 otherwise:

curl localhost:9090/health
# {"status":"ok","checks":{"engine":true,"nats":true}}

GET /ping always returns 200 pong without checking anything, handy as a liveness probe (use /health for readiness):

curl localhost:9090/ping
# pong

🚢 Deployment

NATS Cluster config:

# nats-server.conf
port: 4222
jetstream {
  store_dir: /var/lib/nats
  max_file: 10G
}
cluster {
  name: cosmo-cluster
  listen: 0.0.0.0:6222
  routes: [nats://nats-2:6222, nats://nats-3:6222]
}

Docker Compose:

services:
  nats:
    image: nats:latest
    command: -js -c /etc/nats/nats-server.conf
    volumes:
      - ./nats.conf:/etc/nats/nats-server.conf
      - nats-data:/var/lib/nats

  worker:
    build: .
    environment:
      NATS_URL: nats://nats:4222
    command: bundle exec cosmo -C config/cosmo.yml -c 20 jobs
    deploy:
      replicas: 3

Systemd Service:

# /etc/systemd/system/cosmo.service
[Unit]
Description=Cosmo Background Processor
After=network.target

[Service]
Type=simple
User=deploy
WorkingDirectory=/var/www/myapp
Environment=RAILS_ENV=production
Environment=NATS_URL=nats://localhost:4222
ExecStart=/usr/local/bin/bundle exec cosmo -C config/cosmo.yml -c 20 jobs
Restart=always
RestartSec=10
StandardOutput=syslog
StandardError=syslog
SyslogIdentifier=cosmo

[Install]
WantedBy=multi-user.target

sudo systemctl enable cosmo && sudo systemctl start cosmo

📊 Monitoring

Web UI — mount Cosmo::Web (see Installation) for a live, htmx-powered dashboard with a light, dark, or automatic (system) theme. Fonts and assets are bundled, so it loads nothing from a CDN:

Overview — processed, failed, busy, enqueued, retries, scheduled and dead totals; a 7- or 30-day throughput chart of successful and failed runs; a grid of every worker thread, busy ones colored by stream; the queues; and the jobs running right now, longest first
Jobs — each job queue with its size, busy jobs and redeliveries, paused and resumed in place; then busy, enqueued (per queue), scheduled and dead jobs, with retry and delete for dead ones and their error and backtrace
Processes — every live worker (idle ones included): host, pid, IP, NATS connection name, command line, subscribed streams, busy/concurrency, running/quiet/stopping state, memory, NATS RTT and reconnects. Each worker beats into the _cosmoprocesses KV bucket every 10s, and a crashed worker drops out within 60s
Streams — every stream with its subjects, messages, size, retention and replicas, its consumers' pending, ack-pending and redelivered counts, and pause/resume
Crons — every schedule deployed in NATS, with run-now and delete
Batches — open and finished batches with their progress, succeeded/failed/pending counts and callbacks
Metrics — per job class over today, 7, or 30 days: runs, failures, average execution time (successful runs), and average wait from enqueue to first delivery, plus a per-day chart of runs and average execution time. Recorded by Cosmo::Middleware::Metrics, flushed by every worker's heartbeat into the internal _cosmometrics counters stream; turn it off with config.metrics.enabled = false

Each panel refreshes COSMO_WEB_POLL_INTERVAL seconds after its previous response, never on top of one still running, backs off while refreshes fail, and pauses in a hidden tab. Identical refreshes from several viewers share a single NATS read for COSMO_WEB_CACHE_TTL seconds, and while NATS is unreachable the UI answers at once instead of holding your app server's threads.

Structured logs:

2026-01-23T10:15:30.123Z INFO pid=12345 tid=abc jid=def: start
2026-01-23T10:15:32.456Z INFO pid=12345 tid=abc jid=def elapsed=2.333: done

Stream Metrics:

client = Cosmo::Client.instance
info = client.stream_info("default")

info.state.messages       # Total messages
info.state.bytes          # Total bytes
info.state.consumer_count # Number of consumers

Prometheus — NATS exposes metrics at :8222/metrics:

jetstream_server_store_msgs — Messages in stream
jetstream_consumer_delivered_msgs — Delivered messages
jetstream_consumer_ack_pending — Pending acknowledgments

💼 Examples

Email queue with scheduling:

class EmailJob
  include Cosmo::Job
  options stream: :default, retry: 3

  def perform(user_id, template)
    user = User.find(user_id)
    EmailService.send(user.email, template)
  end
end

EmailJob.perform_async(123, "welcome")
EmailJob.perform_in(1.day, 123, "followup")

Image Processing Pipeline:

class ImageProcessor
  include Cosmo::Stream
  options(
    stream: :images,
    consumer: { subjects: ["images.uploaded.>"] }
  )

  def process_one
    processed = ImageService.process(message.data["url"])
    publish(processed, subject: "images.processed.optimized")
    message.ack
  rescue => e
    logger.error "Processing failed: #{e.message}"
    message.nack(delay: 30_000_000_000) # retry in 30s
  end
end

ImageProcessor.publish({ url: "https://example.com/image.jpg" }, subject: "images.uploaded.user")

Real-Time Analytics:

class AnalyticsAggregator
  include Cosmo::Stream
  options batch_size: 1000, consumer: { subjects: ["events.*.>"] }

  def process(messages)
    aggregates = messages.map(&:data).group_by { |e| e["type"] }.transform_values(&:count)
    Analytics.bulk_insert(aggregates)
    messages.each(&:ack)
  end
end

Made with ❤️ for Ruby

Blast off Cosmonats! 🚀

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
bin		bin
docs		docs
lib		lib
sig		sig
spec		spec
.gitignore		.gitignore
.rspec		.rspec
.rubocop.yml		.rubocop.yml
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
Gemfile		Gemfile
LICENSE.txt		LICENSE.txt
README.md		README.md
Rakefile		Rakefile
config.ru		config.ru
cosmonats.gemspec		cosmonats.gemspec
docker-compose.yml		docker-compose.yml
logo.svg		logo.svg

Latest commit

History

Folders and files

Repository files navigation