#2368 - ChannelBroker updates timeout_at upon contact

- Add `current_delay` into Respondent's sessions, to track the minutes we're currently waiting until the next timeout.
- Fixed a bug in which SMS replies during the last contact attempt refreshed the timeout_at with the fallback_delay instead of the last delay used.
- Improved the test suite by better synchronising the test code with the other processes.
- Deleted best_timeout_option helper and integrated it into consume_retry were it belongs
- Remove ChannelBroker.has_delivery_confirmation? and Runtime.Channel.has_delivery_confirmation? since they could be pattern-matched instead of messaging processes.
- Fixed a bug in the RetriesHistogram that made the last SMS contact appear as the fallback_delay instead of the last delay used.

Fixes #2368

Author: matiasgarciaisaia

lib/ask.ex

@@ -57,7 +57,9 @@ defmodule Ask do
     # See http://elixir-lang.org/docs/stable/elixir/Supervisor.html
     # for other strategies and supported options
     opts = [strategy: :one_for_one, name: Ask.Supervisor]
-    {:ok, _} = Logger.add_backend(Sentry.LoggerBackend)
+    if Mix.env() != :test do
+      {:ok, _} = Logger.add_backend(Sentry.LoggerBackend)
+    end
 
     Supervisor.start_link(children, opts)
   end

lib/ask/channel.ex

@@ -119,6 +119,10 @@ defmodule Ask.Channel do
     end
   end
 
+  # Returns whether the channel sends delivery confirmations callbacks or not.
+  def has_delivery_confirmation?(%{type: "ivr"}), do: false
+  def has_delivery_confirmation?(%{type: "sms"}), do: true
+
   defp validate_patterns(changeset) do
     changeset
     |> validate_patterns_not_empty

lib/ask/retries_histogram.ex

@@ -228,9 +228,12 @@ defmodule Ask.RetriesHistogram do
 
   defp end_flow(survey, flow) do
     fallback_delay = survey |> Survey.fallback_delay() |> minutes_to_hours()
-    [%{type: type}] = Enum.take(flow, -1)
+    [%{type: type, delay: last_delay}] = Enum.take(flow, -1)
 
-    end_flow_delay(type, fallback_delay)
+    # if last SMS delay is 0h, there was no retry - we default to the fallback_delay
+    # See Session.current_timeout/2 for reference
+    delay = if last_delay == 0, do: fallback_delay, else: last_delay
+    end_flow_delay(type, delay)
   end
 
   defp end_flow_delay("ivr", _), do: []

lib/ask/runtime/channel.ex

@@ -3,10 +3,6 @@ defprotocol Ask.Runtime.Channel do
 
   @type t :: map()
 
-  # Returns whether the channel sends delivery confirmations callbacks or not.
-  @spec has_delivery_confirmation?(t()) :: boolean
-  def has_delivery_confirmation?(channel)
-
   # Configure the channel to start communicating with Surveda. For example setup
   # the callback URL on the remote service.
   @spec prepare(t()) :: any()

lib/ask/runtime/channel_broker.ex

@@ -1,9 +1,9 @@
 # NOTE: channels without channel_id (used in some unit tests) share a single process (channel_id: 0)
 defmodule Ask.Runtime.ChannelBroker do
-  alias Ask.Runtime.{ChannelBrokerSupervisor, SurveyLogger}
+  alias Ask.Runtime.{ChannelBrokerSupervisor, Session, SurveyLogger}
   alias Ask.Runtime.ChannelBrokerAgent, as: Agent
   alias Ask.Runtime.ChannelBrokerState, as: State
-  alias Ask.{Channel, Logger, Respondent, Repo, Stats}
+  alias Ask.{Channel, Logger, Respondent, Repo, Stats, SystemTime}
   import Ecto.Query
   use GenServer
 
@@ -32,10 +32,6 @@ defmodule Ask.Runtime.ChannelBroker do
     cast(channel_id, {:setup, channel_type, respondent, token, not_before, not_after})
   end
 
-  def has_delivery_confirmation?(channel_id) do
-    call(channel_id, {:has_delivery_confirmation?})
-  end
-
   def ask(channel_id, channel_type, respondent, token, reply, not_before \\ nil, not_after \\ nil) do
     cast(channel_id, {:ask, channel_type, respondent, token, reply, not_before, not_after})
   end
@@ -286,14 +282,6 @@ defmodule Ask.Runtime.ChannelBroker do
     {:reply, reply, new_state, State.process_timeout(new_state)}
   end
 
-  @impl true
-  def handle_call({:has_delivery_confirmation?}, _from, state) do
-    debug("handle_call[has_delivery_confirmation?]", channel_id: state.channel_id)
-    new_state = refresh_runtime_channel(state)
-    reply = Ask.Runtime.Channel.has_delivery_confirmation?(new_state.runtime_channel)
-    {:reply, reply, new_state, State.process_timeout(new_state)}
-  end
-
   if Mix.env() == :test do
     @impl true
     def handle_call({:has_queued_message?, respondent_id}, _from, state) do
@@ -458,7 +446,7 @@ defmodule Ask.Runtime.ChannelBroker do
   end
 
   defp log_contact(status, respondent) do
-    session = respondent.session |> Ask.Runtime.Session.load()
+    session = respondent.session |> Session.load()
     SurveyLogger.log(
       respondent.survey_id,
       session.flow.mode,
@@ -478,6 +466,8 @@ defmodule Ask.Runtime.ChannelBroker do
       state.runtime_channel
       |> Ask.Runtime.Channel.setup(respondent, token, not_before, not_after)
 
+    update_respondent_timeout(respondent)
+
     case response do
       {:ok, %{verboice_call_id: verboice_call_id}} ->
         channel_state = %{"verboice_call_id" => verboice_call_id}
@@ -501,6 +491,8 @@ defmodule Ask.Runtime.ChannelBroker do
       state.runtime_channel
       |> Ask.Runtime.Channel.ask(respondent, token, reply, state.channel_id)
 
+    update_respondent_timeout(respondent)
+
     case result do
       {:ok, %{nuntium_token: nuntium_token}} ->
         channel_state = %{"nuntium_token" => nuntium_token}
@@ -513,6 +505,16 @@ defmodule Ask.Runtime.ChannelBroker do
     end
   end
 
+  if Mix.env() == :test do
+    defp update_respondent_timeout(%{session: nil} = respondent), do: respondent
+  end
+
+  defp update_respondent_timeout(respondent) do
+    session = respondent.session |> Session.load()
+    timeout_at = Respondent.next_actual_timeout(respondent, session.current_delay, SystemTime.time().now)
+    Respondent.update(respondent, %{timeout_at: timeout_at}, true)
+  end
+
   # Don't schedule automatic GC runs in tests.
   if Mix.env() == :test do
     defp schedule_GC(_), do: nil

lib/ask/runtime/nuntium_channel.ex

@@ -469,7 +469,6 @@ defmodule Ask.Runtime.NuntiumChannel do
 
     def message_inactive?(_, _), do: false
 
-    def has_delivery_confirmation?(_), do: true
     def has_queued_message?(_, _), do: false
     def message_expired?(_, _), do: false
     def cancel_message(_, _), do: :ok

lib/ask/runtime/retries_histogram.ex

@@ -41,7 +41,7 @@ defmodule Ask.Runtime.RetriesHistogram do
       %Session{} = session = Session.load(respondent.session)
 
       session =
-        reallocate_respondent(session, respondent, false, Session.current_timeout(session))
+        reallocate_respondent(session, respondent, false, session.current_delay)
 
       session.respondent
     end
@@ -56,7 +56,7 @@ defmodule Ask.Runtime.RetriesHistogram do
   def retry(session) do
     callback = fn ->
       %Session{respondent: %Respondent{} = respondent} = session
-      reallocate_respondent(session, respondent, ivr?(session), Session.current_timeout(session))
+      reallocate_respondent(session, respondent, ivr?(session), session.current_delay)
     end
 
     run_safe(%{
@@ -146,7 +146,7 @@ defmodule Ask.Runtime.RetriesHistogram do
        ) do
     # sms -> transition to active RetryStat
     if respondent.retry_stat_id,
-      do: reallocate_respondent(session, respondent, false, Session.current_timeout(session))
+      do: reallocate_respondent(session, respondent, false, session.current_delay)
   end
 
   defp do_next_step(_respondent, _session, {:reply, _reply, _}) do

lib/ask/runtime/session.ex

@@ -3,6 +3,7 @@ defmodule Ask.Runtime.Session do
   import Ecto
 
   alias Ask.{
+    Channel,
     Repo,
     QuotaBucket,
     Respondent,
@@ -42,6 +43,7 @@ defmodule Ask.Runtime.Session do
     :flow,
     :respondent,
     :token,
+    :current_delay,
     :fallback_delay,
     :count_partial_results,
     :schedule
@@ -69,16 +71,16 @@ defmodule Ask.Runtime.Session do
       ) do
     flow = Flow.start(questionnaire, mode)
 
+    session_fallback_delay = fallback_delay || Survey.default_fallback_delay()
     session = %Session{
       current_mode: SessionModeProvider.new(mode, channel, retries),
       fallback_mode: SessionModeProvider.new(fallback_mode, fallback_channel, fallback_retries),
       flow: flow,
       respondent: update_section_order(respondent, flow.section_order, persist),
-      fallback_delay: fallback_delay || Survey.default_fallback_delay(),
+      fallback_delay: session_fallback_delay,
       count_partial_results: count_partial_results,
       schedule: schedule
     }
-
     run_flow(session, persist)
   end
 
@@ -105,25 +107,23 @@ defmodule Ask.Runtime.Session do
         {:ok, session, %Reply{}, base_timeout(session) + current_timeout(session)}
 
       true ->
-        timeout(session, nil)
+        do_timeout(session)
     end
   end
 
-  def timeout(%{current_mode: %{retries: []}, fallback_mode: nil} = session, _) do
+  defp do_timeout(%{current_mode: %{retries: []}, fallback_mode: nil} = session) do
     session = %{session | respondent: RetriesHistogram.remove_respondent(session.respondent)}
     terminate(session)
   end
 
-  def timeout(%{current_mode: %{retries: []}} = session, _) do
+  defp do_timeout(%{current_mode: %{retries: []}} = session) do
     switch_to_fallback_mode(session)
   end
 
-  def timeout(%Session{} = session, _) do
-    best_timeout_option = best_timeout_option(session)
+  defp do_timeout(%Session{} = session) do
     session = retry(session)
 
-    # The new session will timeout as defined by hd(retries)
-    {:ok, session, %Reply{}, best_timeout_option || current_timeout(session)}
+    {:ok, session, %Reply{}, session.current_delay}
   end
 
   @doc """
@@ -215,11 +215,11 @@ defmodule Ask.Runtime.Session do
     Respondent.update(respondent, %{section_order: section_order}, persist)
   end
 
-  def current_timeout(%Session{current_mode: %{retries: []}, fallback_delay: fallback_delay}) do
+  defp current_timeout(%Session{current_mode: %{retries: []}, fallback_delay: fallback_delay}) do
     fallback_delay
   end
 
-  def current_timeout(%Session{current_mode: %{retries: [next_retry | _]}}) do
+  defp current_timeout(%Session{current_mode: %{retries: [next_retry | _]}}) do
     next_retry
   end
 
@@ -335,6 +335,7 @@ defmodule Ask.Runtime.Session do
       flow: session.flow |> Flow.dump(),
       respondent_id: session.respondent.id,
       token: session.token,
+      current_delay: session.current_delay,
       fallback_delay: session.fallback_delay,
       count_partial_results: session.count_partial_results,
       schedule: session.schedule |> Schedule.dump!()
@@ -348,6 +349,7 @@ defmodule Ask.Runtime.Session do
       flow: Flow.load(state["flow"]),
       respondent: Repo.get(Ask.Respondent, state["respondent_id"]),
       token: state["token"],
+      current_delay: state["current_delay"],
       fallback_delay: state["fallback_delay"],
       count_partial_results: state["count_partial_results"],
       schedule: state["schedule"] |> Schedule.load!()
@@ -601,6 +603,14 @@ defmodule Ask.Runtime.Session do
       |> add_mode_attempt.()
 
     mode_start(session)
+    |> update_current_delay
+  end
+
+  defp update_current_delay({:ok, session, reply, timeout}) do
+   {:ok, %{session | current_delay: timeout}, reply, timeout}
+  end
+  defp update_current_delay({:end, _, _} = flow_result) do
+    flow_result
   end
 
   defp apply_patterns_if_match(patterns, respondent, persist) do
@@ -628,7 +638,7 @@ defmodule Ask.Runtime.Session do
   defp log_prompts(reply, channel, mode, respondent, force \\ false, persist \\ true) do
     if persist do
       if force ||
-           !ChannelBroker.has_delivery_confirmation?(channel.id) do
+           !Channel.has_delivery_confirmation?(channel) do
         disposition = Reply.disposition(reply) || respondent.disposition
 
         Enum.each(Reply.steps(reply), fn step ->
@@ -712,12 +722,6 @@ defmodule Ask.Runtime.Session do
     %{session | token: nil}
   end
 
-  defp best_timeout_option(%{current_mode: %{retries: retries}, fallback_mode: nil})
-       when length(retries) == 1,
-       do: hd(retries)
-
-  defp best_timeout_option(_), do: nil
-
   defp terminate(%{current_mode: %SMSMode{}, respondent: respondent}) do
     {:failed, respondent}
   end
@@ -757,12 +761,19 @@ defmodule Ask.Runtime.Session do
     result
   end
 
+  defp consume_retry(%{current_mode: %{retries: [retry]}, fallback_mode: nil} = session) do
+    %{session | current_mode: %{session.current_mode | retries: []}, current_delay: retry}
+  end
+
   defp consume_retry(%{current_mode: %{retries: [_ | retries]}} = session) do
-    %{session | current_mode: %{session.current_mode | retries: retries}}
+    session = %{session | current_mode: %{session.current_mode | retries: retries}}
+    current_delay = current_timeout(session)
+    %{session | current_delay: current_delay}
   end
 
   defp consume_retry(%{current_mode: %{retries: []}} = session) do
-    session
+    current_delay = current_timeout(session)
+    %{session | current_delay: current_delay}
   end
 
   defp add_session_mode_attempt!(%Session{} = session),
@@ -831,7 +842,7 @@ defmodule Ask.Runtime.Session do
           persist
         )
 
-        {:ok, %{session | flow: flow}, reply, current_timeout(session)}
+        {:ok, %{session | flow: flow}, reply, session.current_delay}
     end
   end
 
@@ -841,7 +852,7 @@ defmodule Ask.Runtime.Session do
         {:failed, session.respondent}
 
       _ ->
-        {:hangup, %{session | flow: flow}, reply, current_timeout(session), session.respondent}
+        {:hangup, %{session | flow: flow}, reply, session.current_delay, session.respondent}
     end
   end
 

lib/ask/runtime/survey_broker.ex

@@ -223,22 +223,6 @@ defmodule Ask.Runtime.SurveyBroker do
     )
   end
 
-  def recontact_queued_respondents(respondent_ids) do
-    # We're recontacting respondants that were queued
-    # in channel broker after it fails
-    Repo.all(
-      from r in Respondent,
-        select: r.id,
-        where: r.id in ^respondent_ids
-    )
-    |> Enum.each(fn respondent_id ->
-      respondent = Respondent |> Repo.get(respondent_id)
-      session = respondent.session |> Session.load()
-      # We have loaded everything neccesary, now contact them without consuming a retry
-      Ask.Runtime.Session.contact_respondent(session)
-    end)
-  end
-
   defp retry_respondents(now) do
     # Select projects that have respondents to retry, then retry them respecting the project's batch limit
     Repo.all(

lib/ask/runtime/verboice_channel.ex

@@ -479,7 +479,6 @@ defmodule Ask.Runtime.VerboiceChannel do
   def check_status(_), do: :up
 
   defimpl Ask.Runtime.Channel, for: Ask.Runtime.VerboiceChannel do
-    def has_delivery_confirmation?(_), do: false
     def ask(_, _, _, _, _), do: throw(:not_implemented)
     def prepare(_), do: :ok
     def messages_count(_, _, _, _, _), do: 1