mule/runtime/drainer

Synchronous queue draining, mirroring Oban.Queue.Drainer (queue/drainer.ex): stage scheduled rows when asked, fetch the queue’s available jobs through the engine, execute each in the CALLING process with the same executor regular dispatch uses (acks are real — a failure lands retryable with genuine backoff), and count terminal states.

Two Elixir-inherited details the port makes deliberate choices on:

• The fetch meta is built literally rather than through init_queue: the port’s init merges the persisted pause/scale overlay (parity §9), which would make draining a runtime-paused queue silently fetch nothing — and “unpausing” via put_meta would write through to the real overlay. The literal meta reproduces what Basic’s init gives the Elixir drainer (paused: false).
• Elixir’s Process.put(:mule_draining, true) is not ported: nothing in Oban 2.22’s lib/ reads it (a Pro hook), so there is no behavior to mirror.

Hazards, both true in Elixir too: a worker timeout arms timer:exit_after in the calling process, so a timed-out drained job kills the drain caller; and WithRecursion(True) plus scheduled staging with an always-snoozing worker never terminates — each pass re-stages the snoozed row and drains it again, changing the counts every time.

pub type DrainOption {
  WithLimit(limit: Int)
  WithRecursion(recursion: Bool)
  WithSafety(safety: Bool)
  WithScheduled(scheduled: Bool)
  WithScheduledBefore(scheduled_before: timestamp.Timestamp)
}

pub type DrainResult {
  DrainResult(
    cancelled: Int,
    discard: Int,
    failure: Int,
    snoozed: Int,
    success: Int,
  )
}

pub fn drain(
  config config: config.Config,
  queue queue: String,
  options options: List(DrainOption),
) -> Result(DrainResult, engine.EngineError)