Recording to S3

Store recorded output in S3, or an S3-compatible service such as Cloudflare R2 or Tigris. This is useful for large results and output that should follow the bucket's retention policy rather than remain in Postgres.

This backend uses Req with its SigV4 support. Add Req and aws_signature:

def deps do
  [
    {:req, "~> 0.5"},
    {:aws_signature, "~> 0.3"}
  ]
end

The Module

The module reads credentials from the environment at runtime, while non-sensitive options such as the bucket and region may be configured on workers. It disables response decoding because recorded payloads are opaque binaries.

defmodule MyApp.S3 do
  @behaviour Oban.Pro.Storage

  @impl Oban.Pro.Storage
  def init(opts) do
    bucket = Keyword.get(opts, :bucket, "my-bucket")
    region = Keyword.get(opts, :region, "us-east-1")

    Req.new(
      base_url: "https://#{bucket}.s3.#{region}.amazonaws.com",
      decode_body: false,
      aws_sigv4: [
        service: "s3",
        region: region,
        access_key_id: System.fetch_env!("AWS_ACCESS_KEY_ID"),
        secret_access_key: System.fetch_env!("AWS_SECRET_ACCESS_KEY")
      ]
    )
  end

  @impl Oban.Pro.Storage
  def put(key, payload, req) do
    case Req.put(req, url: "/#{key}", body: payload) do
      {:ok, %{status: status}} when status in 200..299 -> :ok
      {:ok, response} -> {:error, response}
      {:error, reason} -> {:error, reason}
    end
  end

  @impl Oban.Pro.Storage
  def fetch_all(keys, req) do
    payloads =
      for {:ok, {key, body}} <- Task.async_stream(keys, &fetch_one(&1, req), ordered: false),
          not is_nil(body),
          into: %{},
          do: {key, body}

    {:ok, payloads}
  end

  @impl Oban.Pro.Storage
  def delete_all(keys, req) do
    Enum.each(keys, &Req.delete(req, url: "/#{&1}"))
  end

  defp fetch_one(key, req) do
    case Req.get(req, url: "/#{key}") do
      {:ok, %{status: 200, body: body}} -> {key, body}
      _other -> {key, nil}
    end
  end
end

The recipe fetches batches concurrently and omits missing objects. To keep output after its job is pruned, make delete_all/2 a no-op and use bucket lifecycle rules for retention.

Any S3-compatible store works the same way by changing the base_url in init/1. Cloudflare R2 (https://<account_id>.r2.cloudflarestorage.com/<bucket>) and Tigris (https://fly.storage.tigris.dev) both sign with region: "auto".

Using It

Configure the backend on a recorded worker and pass non-sensitive options such as the bucket:

use Oban.Pro.Worker, recorded: [storage: {MyApp.S3, bucket: "oban-output"}]

To send every worker's output to S3 without annotating each one, set it as the compile-time default:

config :oban_pro, Oban.Pro.Worker, recorded: [storage: MyApp.S3]