Fleet

High-throughput, zero-bloat background job processing engine for Elixir, powered by PgQue and Broadway.

Key Highlights

Installation

Add fleet to your list of dependencies in mix.exs:

def deps do
  [
    {:fleet, "~> 0.1.0"}
  ]
end

1. Run the Migration

Generate a migration in your application:

mix ecto.gen.migration add_fleet

And invoke Fleet.Migration:

defmodule MyApp.Repo.Migrations.AddFleet do
  use Ecto.Migration

  def up do
    Fleet.Migration.up()
  end

  def down do
    Fleet.Migration.down()
  end
end

Run mix ecto.migrate to create the schema and functions.

Defining Workers

Define background workers by using Fleet.Worker:

defmodule MyApp.Workers.SendEmail do
  use Fleet.Worker, queue: :mailers, max_attempts: 5

  @impl Fleet.Worker
  def perform(%Fleet.Job{args: %{"to" => to, "subject" => subject}}) do
    MyApp.Mailer.send(to, subject)
  end
end

Strongly Typed Workers with Fleet.Schema

For compile-time type safety, automatic runtime casting, and changeset validation, use Fleet.Schema:

defmodule MyApp.Workers.SendInvoice do
  use Fleet.Worker, queue: :billing, max_attempts: 3
  use Fleet.Schema

  embedded_schema do
    field :invoice_id, :integer
    field :amount, :decimal
    field :currency, :string, default: "USD"
  end

  # Optional: custom validation rules
  def changeset(schema, attrs) do
    schema
    |> cast(attrs, [:invoice_id, :amount, :currency])
    |> validate_required([:invoice_id, :amount])
  end

  @impl Fleet.Worker
  def perform(%Fleet.Job{args: %__MODULE__{invoice_id: id, amount: amount, currency: curr}}) do
    # 100% typed: `amount` is Decimal, `id` is integer, `curr` defaults to "USD"
    MyApp.Billing.charge(id, amount, curr)
  end
end

Enqueueing Typed Jobs

# Auto-casts strings to Decimals, integers, DateTimes, etc. Raises on invalid changeset
job = MyApp.Workers.SendInvoice.new(%{invoice_id: 101, amount: "49.99"})
Fleet.insert(MyApp.Repo, job)

# Safe constructor returning {:ok, job} | {:error, changeset} (ideal for web controllers)
case MyApp.Workers.SendInvoice.build(params) do
  {:ok, job} -> Fleet.insert(MyApp.Repo, job)
  {:error, %Ecto.Changeset{} = cs} -> {:error, cs}
end

Snoozing and Retries

@impl Fleet.Worker
def perform(%Fleet.Job{args: %{"task_id" => id}}) do
  if MyApp.TaskService.ready?(id) do
    MyApp.TaskService.run(id)
  else
    # Delay for 60 seconds
    {:snooze, 60}
  end
end

Batch Workers (Chunks)

Process multiple jobs together in a single batch, automatically buffered by Broadway according to batch_size and batch_timeout:

defmodule MyApp.Workers.BulkEvents do
  use Fleet.Worker, queue: :events

  @impl Fleet.Worker
  def perform(_job), do: :ok

  @impl Fleet.Worker
  def perform_batch(jobs) do
    payloads = Enum.map(jobs, fn job -> job.args end)
    MyApp.Analytics.send_bulk(payloads)
    :ok
  end
end

High-Performance Database Ingestion (with ecto_unnest)

When your batch worker needs to bulk-insert processed jobs into your application's Ecto schemas, pairing perform_batch/1 with ecto_unnest produces a zero-bloat, constant-SQL ingestion pipeline:

defmodule MyApp.Workers.ImportUsers do
  use Fleet.Worker, queue: :imports

  @impl Fleet.Worker
  def perform_batch(jobs) do
    # Extract records buffered by Broadway
    user_ids = Enum.map(jobs, & &1.args["user_id"])
    emails   = Enum.map(jobs, & &1.args["email"])
    names    = Enum.map(jobs, & &1.args["name"])

    # Constant SQL bulk insert using PostgreSQL unnest()
    EctoUnnest.insert_all(
      MyApp.Repo,
      MyApp.User,
      %{user_id: user_ids, email: emails, name: names},
      placeholders: %{inserted_at: DateTime.utc_now()}
    )

    :ok
  end
end

Broadway provides the memory protection and backpressure, while unnest() avoids prepared statement proliferation and eliminates SQL parameter limit bottlenecks.

Enqueueing Jobs

Direct Insertion

%{to: "jane@example.com", subject: "Welcome"}
|> MyApp.Workers.SendEmail.new()
|> Fleet.insert(MyApp.Repo)

Transactional Insertion with Ecto.Multi

Ensure jobs are only enqueued if your business transaction commits:

Ecto.Multi.new()
|> Ecto.Multi.insert(:user, user_changeset)
|> Fleet.insert(:welcome_email, fn %{user: user} ->
  MyApp.Workers.SendEmail.new(%{"to" => user.email, "subject" => "Welcome!"})
end)
|> MyApp.Repo.transaction()

Bulk Enqueueing (Native PostgreSQL unnest Batching)

Enqueue hundreds or thousands of jobs in a single round-trip:

jobs = Enum.map(users, fn user ->
  MyApp.Workers.SendEmail.new(%{"to" => user.email})
end)

Fleet.insert_all(MyApp.Repo, jobs)

Tip

Zero Statement Bloat: Under the hood, Fleet.insert_all/2 delegates to pgque.send_batch/3, which uses PostgreSQL's native unnest($2::text[]) WITH ORDINALITY. The query text remains 100% constant regardless of batch size, making it fully compatible with transaction-mode connection poolers like PgBouncer and completely immune to PostgreSQL's 65,535 query parameter cap.

Configuration and Supervision

Add Fleet to your application supervision tree in lib/my_app/application.ex:

def start(_type, _args) do
  children = [
    MyApp.Repo,
    {Fleet,
     repo: MyApp.Repo,
     run_ticker: true,
     queues: [
       default: [concurrency: 10],
       mailers: [concurrency: 20, poll_interval: 50],
       events: [concurrency: 5, batch_size: 50, batch_timeout: 1000]
     ],
     cron: [
       {"@hourly", MyApp.Workers.CleanupWorker},
       {"@every 30s", MyApp.Workers.HeartbeatWorker, args: %{"scope" => "all"}}
     ]}
  ]

  opts = [strategy: :one_for_one, name: MyApp.Supervisor]
  Supervisor.start_link(children, opts)
end

Recipes & Architectural Patterns

Recipe 1: Zero-Bloat Recurring & Mid-Month Billing (The Reconciliation Sweep)

A common challenge in subscription systems is billing customers periodically (e.g., monthly) while supporting signups that occur on arbitrary days throughout the month.

Warning

Anti-Pattern: 30-Day Sleeping Queue Jobs Enqueueing a background job scheduled 30 days in the future for every subscriber is an architectural anti-pattern:

  1. Dead-Tuple Bloat: Delayed jobs sitting in database tables cannot leverage rotating table TRUNCATE operations; they accumulate dead tuples on DELETE and cause index bloat on scheduled timestamps.
  2. Stale / Zombie State: If a subscriber cancels, upgrades, or updates their credit card on day 15, an enqueued job becomes a "zombie" that must either be manually searched and deleted or defended with complex pre-execution checks.

The Recommended Pattern: State in Postgres + Scheduled Sweep

Instead of storing scheduling state in queue tables, store the next renewal timestamp in your application database (subscriptions.next_billing_at), index it with a partial index, and run an @hourly or @daily reconciliation sweep using Fleet.Cron.

1. Database Schema with Partial Index

In your PostgreSQL migration:

create table(:subscriptions) do
  add :user_id, references(:users), null: false
  add :status, :string, default: "active", null: false
  add :next_billing_at, :utc_datetime, null: false
  add :amount, :decimal, precision: 10, scale: 2, null: false
  timestamps()
end

# Ultra-compact, fast partial index:
create index(:subscriptions, [:next_billing_at], where: "status = 'active'")
2. Handling Mid-Month Signups

When a user signs up on any day of the month (e.g. October 14th), charge their initial invoice immediately and schedule their next_billing_at 30 days into the future within the same transaction:

def register_and_subscribe(user_params, card_token) do
  Ecto.Multi.new()
  |> Ecto.Multi.insert(:user, User.changeset(user_params))
  |> Ecto.Multi.insert(:subscription, fn %{user: user} ->
    Subscription.changeset(%{
      user_id: user.id,
      status: "active",
      next_billing_at: DateTime.add(DateTime.utc_now(), 30, :day)
    })
  end)
  # Enqueue immediate first charge (or process synchronously):
  |> Fleet.insert(:initial_charge, fn %{subscription: sub} ->
    MyApp.Workers.ChargeSubscription.new(%{"subscription_id" => sub.id, "initial" => true})
  end)
  |> MyApp.Repo.transaction()
end
3. The Hourly Reconciliation Sweep Worker

Use Fleet.Cron in your application supervision tree:

# lib/my_app/application.ex
{Fleet,
 repo: MyApp.Repo,
 cron: [
   {"@hourly", MyApp.Workers.BillingSweepWorker}
 ]}

The sweep queries only active subscriptions due right now and enqueues instant jobs in bulk:

defmodule MyApp.Workers.BillingSweepWorker do
  use Fleet.Worker, queue: :billing

  import Ecto.Query

  @impl Fleet.Worker
  def perform(_job) do
    now = DateTime.utc_now()

    # Query only active subscriptions that are due
    due_subs =
      MyApp.Repo.all(
        from s in MyApp.Subscription,
          where: s.status == "active" and s.next_billing_at <= ^now,
          limit: 1000
      )

    # Bulk-enqueue immediate jobs with constant SQL unnest
    jobs =
      Enum.map(due_subs, fn sub ->
        MyApp.Workers.ChargeSubscription.new(%{"subscription_id" => sub.id})
      end)

    {:ok, _} = Fleet.insert_all(MyApp.Repo, jobs)
    :ok
  end
end
4. The Charge Worker

The worker executes under Broadway's concurrency control, charges the payment gateway, and advances next_billing_at:

defmodule MyApp.Workers.ChargeSubscription do
  use Fleet.Worker, queue: :billing, max_attempts: 3

  @impl Fleet.Worker
  def perform(%Fleet.Job{args: %{"subscription_id" => sub_id}}) do
    case MyApp.Billing.charge_and_renew(sub_id) do
      :ok ->
        :ok

      {:error, :card_declined} ->
        # Mark subscription past_due; do not retry
        MyApp.Billing.mark_past_due(sub_id)
        :ok

      {:error, :gateway_timeout} ->
        # Transient failure: Fleet automatically retries with exponential backoff
        {:error, :gateway_timeout}
    end
  end
end

Why this preserves Fleet's Zero-Bloat guarantees:

Ring-Buffer Retention & Historical Queries

Fleet retains completed jobs in circular partition tables according to each queue's configured retention. You can query past jobs directly with Fleet.History:

# Count jobs created in the last 24 hours
since = DateTime.add(DateTime.utc_now(), -86400, :second)
{:ok, count} = Fleet.History.count_jobs(MyApp.Repo, :mailers, since: since)

# Paginate recent historical events
{:ok, recent_jobs} = Fleet.History.list_jobs(MyApp.Repo, :mailers, limit: 25, offset: 0)

# Fetch a single historical job by ID
{:ok, job} = Fleet.History.get_job(MyApp.Repo, :mailers, 1024)

Telemetry Events

Fleet emits standard :telemetry events:

License

Apache License 2.0.