Elixir技术问题：替换预测拨号器后实现任务失败断点重试

Hey there! Awesome to hear your Elixir predictive dialer is outperforming expectations—great work so far! Let’s work through implementing that breakpoint retry feature you need, so you can pick up right where tasks failed. Here’s a practical, Elixir-native approach tailored to your code:

The key to breakpoint retry is tracking the execution state of each contact (pending, in progress, succeeded, failed) and filtering out completed tasks when restarting. We’ll use Elixir’s built-in tools like Task.Supervisor for task management and ETS (or a database for persistent storage) to track progress.

1. Set Up Progress Tracking #

First, create a module to handle tracking contact states. We’ll use an ETS table for in-memory tracking (swap this with an Ecto schema if you need persistence across app restarts):

defmodule Dialer.ProgressTracker do
  @doc """
  Initialize the ETS table when the app starts (call this in your application's start callback)
  """
  def init do
    :ets.new(:dialer_task_progress, [:named_table, :public, :set])
  end

  def mark_started(contact_id), do: :ets.insert(:dialer_task_progress, {contact_id, :started})
  def mark_success(contact_id), do: :ets.insert(:dialer_task_progress, {contact_id, :success})
  def mark_failed(contact_id), do: :ets.insert(:dialer_task_progress, {contact_id, :failed})

  @doc """
  Filter contacts to only those that haven't succeeded or need retries
  """
  def get_pending_contacts(contacts) do
    Enum.filter(contacts, fn contact ->
      case :ets.lookup(:dialer_task_progress, contact.id) do
        [] -> true          # Never processed
        [{_, :failed}] -> true # Failed, needs retry
        _ -> false          # Already succeeded or in progress
      end
    end)
  end
end

2. Use Task Supervisor for Failure Handling #

Replace raw Task.start_link with a Task.Supervisor to safely monitor tasks and capture failures. Define the supervisor in your application’s supervision tree:

# In your application.ex start callback
children = [
  # ... other children
  {Task.Supervisor, name: Dialer.TaskSupervisor}
]

3. Modify Your Perform Function to Respect Progress #

Update your existing perform function to filter pending contacts, track state, and handle failures:

def perform(phonebook_contacts, account_number, ...) do
  alias FSModEvent.Connection, as: C
  alias Dialer.ProgressTracker

  # Only process contacts that need work (pending or failed)
  pending_contacts = ProgressTracker.get_pending_contacts(phonebook_contacts)

  # Use chunking as before, but wrap tasks in error handling
  for chunk <- Enum.chunk(pending_contacts, 100, 100, []), contact <- chunk do
    unless Telephony.user_balance(account_number) <= 0 do
      Task.Supervisor.start_child(Dialer.TaskSupervisor, fn ->
        ProgressTracker.mark_started(contact.id)
        try do
          # Your existing dialing logic goes here
          dial_contact(contact)
          ProgressTracker.mark_success(contact.id)
        rescue
          error ->
            ProgressTracker.mark_failed(contact.id)
            Logger.error("Dial failed for contact #{contact.id}: #{inspect(error)}")
        end
      end)
    end
  end
end

# Helper for your dialing logic (replace with your actual code)
defp dial_contact(contact) do
  # ... connect to FSModEvent, initiate call, etc.
end

4. Resume from Breakpoints #

To resume after a failure or app restart, simply call perform/... again. The get_pending_contacts/1 function will automatically skip contacts that already succeeded, and only retry failed or unprocessed ones.

Bonus: Add Retry Limits #

Avoid infinite retries by tracking retry counts. Update the progress tracker to include retry attempts:

defmodule Dialer.ProgressTracker do
  # ... existing functions

  def mark_failed(contact_id) do
    case :ets.lookup(:dialer_task_progress, contact_id) do
      [{_, :failed, retry_count}] when retry_count >= 3 ->
        :ets.insert(:dialer_task_progress, {contact_id, :abandoned})
      [{_, :failed, retry_count}] ->
        :ets.insert(:dialer_task_progress, {contact_id, :failed, retry_count + 1})
      _ ->
        :ets.insert(:dialer_task_progress, {contact_id, :failed, 1})
    end
  end

  def get_pending_contacts(contacts) do
    Enum.filter(contacts, fn contact ->
      case :ets.lookup(:dialer_task_progress, contact.id) do
        [] -> true
        [{_, :failed, count}] when count < 3 -> true
        _ -> false
      end
    end)
  end
end

This way, contacts will only be retried up to 3 times before being marked as abandoned.

内容的提问来源于stack exchange，提问作者devie