lua-patterns.md

Lua Hook Patterns

Copy-paste patterns for common scheduling scenarios. See concepts for hook API details.

Tenant fairness

Assign each tenant its own fairness group so the DRR scheduler gives equal delivery bandwidth:

function on_enqueue(msg)
  return {
    fairness_key = msg.headers["tenant_id"]   -- nil: the message joins the unkeyed group
  }
end

With weighted tiers

Premium tenants get more bandwidth:

function on_enqueue(msg)
  local tier = msg.headers["tier"] or "standard"
  local weight = 1
  if tier == "premium" then weight = 3 end
  if tier == "enterprise" then weight = 5 end

  return {
    fairness_key = msg.headers["tenant_id"],
    weight = weight
  }
end

With dynamic weights from config

function on_enqueue(msg)
  local tenant = msg.headers["tenant_id"] or "default"
  local weight = tonumber(fila.get("weight:" .. tenant) or "1")

  return {
    fairness_key = tenant,
    weight = weight
  }
end

Set weights at runtime: fila config set weight:acme 5

---

Throttle keys

Throttles are declared by consumers (see throttling.md). Lua's role is computing the values a throttle is keyed by, returned as attributes from on_enqueue, when they cannot simply be read from a header or the fairness key — including setting an attribute only on the messages a throttle should apply to.

Attributes are computed once, at enqueue, and stored with the message. Changing the script affects messages enqueued afterwards.

Derived customer account

Several customer IDs map to one billing account, and the downstream limit is per account:

function on_enqueue(msg)
  local customer = msg.headers["customer"]
  if not customer then
    return {}   -- attribute absent: the throttle's missing-key policy applies
  end
  return { attributes = { account = fila.get("account:" .. customer) or customer } }
end

consumer
    .subscribe("charges")
    .throttle(
        Throttle::named("stripe-per-account")
            .key([Key::attribute("account")])
            .limit(10, Duration::from_secs(1)),
    )
    .await?;

Region from a composite header

function on_enqueue(msg)
  -- "eu-west-1:acme" -> "eu-west-1"
  local target = msg.headers["target"] or ""
  return { attributes = { region = target:match("^([^:]+)") } }
end

A nil value means the attribute is absent, and the throttle's missing-key policy applies.

---

Exponential backoff retry

A fixed attempt limit with exponential backoff doesn't need a script — it's what the queue's retry policy does by default. Use on_failure when the decision depends on the message, the error, or runtime config. When the script runs, its decision is final: max_attempts does not limit it.

Retry with increasing delays, dead-letter after max attempts:

function on_failure(msg)
  if msg.attempts >= 5 then
    return { action = "dlq" }
  end

  -- 1s, 2s, 4s, 8s, 16s
  local delay = math.min(1000 * (2 ^ (msg.attempts - 1)), 60000)
  return { action = "retry", delay_ms = delay }
end

With configurable max retries

function on_failure(msg)
  local max = tonumber(fila.get("max_retries") or "5")
  if msg.attempts >= max then
    return { action = "dlq" }
  end

  local delay = math.min(1000 * (2 ^ (msg.attempts - 1)), 60000)
  return { action = "retry", delay_ms = delay }
end

Change at runtime: fila config set max_retries 10

Linear backoff

function on_failure(msg)
  if msg.attempts >= 5 then
    return { action = "dlq" }
  end

  -- 5s, 10s, 15s, 20s, 25s
  return { action = "retry", delay_ms = 5000 * msg.attempts }
end

Immediate retry (no delay)

function on_failure(msg)
  if msg.attempts >= 3 then
    return { action = "dlq" }
  end
  return { action = "retry", delay_ms = 0 }
end

---

Header-based routing

Use headers to make dynamic scheduling decisions.

Route by priority

function on_enqueue(msg)
  local priority = msg.headers["priority"] or "normal"
  local weights = {
    critical = 10,
    high = 5,
    normal = 2,
    low = 1
  }

  return {
    fairness_key = "priority:" .. priority,
    weight = weights[priority] or 2
  }
end

Route by region

function on_enqueue(msg)
  local region = msg.headers["region"] or "default"

  return {
    fairness_key = "region:" .. region
  }
end

Worker crashes vs. errors

An expired lease reaches on_failure with msg.reason = "lease_expired". A message that keeps crashing workers is likely poison; one that returns errors may just be waiting on a dependency:

function on_failure(msg)
  if msg.reason == "lease_expired" and msg.attempts >= 2 then
    return { action = "dlq" }
  end
  if msg.attempts >= 10 then
    return { action = "dlq" }
  end
  return { action = "retry", delay_ms = math.min(1000 * (2 ^ (msg.attempts - 1)), 60000) }
end

Conditional dead-letter by error type

function on_failure(msg)
  -- Permanent errors: dead-letter immediately
  if msg.error:find("4%d%d") then  -- HTTP 4xx
    return { action = "dlq" }
  end

  -- Transient errors: retry with backoff
  if msg.attempts >= 5 then
    return { action = "dlq" }
  end

  local delay = 1000 * (2 ^ (msg.attempts - 1))
  return { action = "retry", delay_ms = delay }
end

Feature flag gating

function on_enqueue(msg)
  local tenant = msg.headers["tenant"] or "default"
  local new_flow = fila.get("feature:new_flow:" .. tenant)

  if new_flow == "enabled" then
    return { fairness_key = tenant .. ":v2", weight = 1 }
  end

  return { fairness_key = tenant, weight = 1 }
end

# Enable new flow for one tenant
fila config set feature:new_flow:acme enabled