| 1 |
|
defmodule WalletSettlement.QueueConfig do |
| 2 |
|
@moduledoc """ |
| 3 |
|
Oban queue topology and priority configuration for async settlement processing. |
| 4 |
|
|
| 5 |
|
Queue classes (criticality-based isolation as per ADR 0002 outbox/inbox pattern): |
| 6 |
|
|
| 7 |
|
| Queue | Priority | Max Concurrency | Max Attempts | Purpose | |
| 8 |
|
|---------------------|----------|-----------------|--------------|--------------------------------------| |
| 9 |
|
| :settlement_high | 1 | 5 | 3 | Settlement batch execution | |
| 10 |
|
| :settlement_normal | 2 | 10 | 5 | Reconciliation runs | |
| 11 |
|
| :settlement_low | 3 | 20 | 3 | Exception status updates / exports | |
| 12 |
|
| :notifications_high | 1 | 10 | 3 | Transactional alerts (send-critical) | |
| 13 |
|
| :notifications_low | 3 | 30 | 5 | Marketing / digest notifications | |
| 14 |
|
|
| 15 |
|
Retry/backoff policy: |
| 16 |
|
- Retries use exponential backoff: base_sleep_time = 15_000 + 2^attempt * 1000 ms. |
| 17 |
|
- Poison-message (max_attempts exceeded): job moved to :discarded state with final error. |
| 18 |
|
- Dead-letter escalation criteria: >10 discarded jobs per queue in any 5-minute window |
| 19 |
|
triggers PagerDuty alert and auto-pauses the queue. |
| 20 |
|
|
| 21 |
|
In production, configure Oban in `config/runtime.exs`: |
| 22 |
|
|
| 23 |
|
config :oban, Oban, |
| 24 |
|
repo: WalletWeb.Repo, |
| 25 |
|
plugins: [ |
| 26 |
|
{Oban.Plugins.Pruner, max_age: 86_400}, |
| 27 |
|
{Oban.Plugins.Stager, interval: 1_000}, |
| 28 |
|
{Oban.Plugins.Lifeline, rescue_after: :timer.minutes(30)}, |
| 29 |
|
{Oban.Plugins.Gossip, interval: :timer.seconds(30)} |
| 30 |
|
], |
| 31 |
|
queues: [ |
| 32 |
|
settlement_high: [limit: 5, max_attempts: 3], |
| 33 |
|
settlement_normal: [limit: 10, max_attempts: 5], |
| 34 |
|
settlement_low: [limit: 20, max_attempts: 3], |
| 35 |
|
notifications_high: [limit: 10, max_attempts: 3], |
| 36 |
|
notifications_low: [limit: 30, max_attempts: 5] |
| 37 |
|
] |
| 38 |
|
|
| 39 |
|
For CI (no database), workers are invoked directly via `perform/1`. |
| 40 |
|
""" |
| 41 |
|
|
| 42 |
|
@queues %{ |
| 43 |
|
settlement_high: %{priority: 1, max_concurrency: 5, max_attempts: 3}, |
| 44 |
|
settlement_normal: %{priority: 2, max_concurrency: 10, max_attempts: 5}, |
| 45 |
|
settlement_low: %{priority: 3, max_concurrency: 20, max_attempts: 3}, |
| 46 |
|
notifications_high: %{priority: 1, max_concurrency: 10, max_attempts: 3}, |
| 47 |
|
notifications_low: %{priority: 3, max_concurrency: 30, max_attempts: 5} |
| 48 |
|
} |
| 49 |
|
|
| 50 |
|
@doc "Returns the full queue topology configuration map." |
| 51 |
:-( |
def queues, do: @queues |
| 52 |
|
|
| 53 |
|
@doc "Returns config for a named queue, or `{:error, :unknown_queue}`." |
| 54 |
|
def queue(name) do |
| 55 |
13 |
case Map.fetch(@queues, name) do |
| 56 |
13 |
{:ok, cfg} -> {:ok, cfg} |
| 57 |
:-( |
:error -> {:error, :unknown_queue} |
| 58 |
|
end |
| 59 |
|
end |
| 60 |
|
|
| 61 |
|
@doc """ |
| 62 |
|
Calculates exponential backoff sleep (ms) for a given attempt number (1-based). |
| 63 |
|
|
| 64 |
|
Formula: 15_000 + 2^attempt * 1_000 ms |
| 65 |
|
""" |
| 66 |
|
@spec backoff_ms(attempt :: pos_integer()) :: non_neg_integer() |
| 67 |
|
def backoff_ms(attempt) when is_integer(attempt) and attempt >= 1 do |
| 68 |
4 |
15_000 + :math.pow(2, attempt) * 1_000 |> round() |
| 69 |
|
end |
| 70 |
|
end |