Forward data received on a failed pair and re-schedule this pair (#53)

Author: mickel8

lib/ex_ice/priv/conn_check_handler/controlled.ex

@@ -19,9 +19,14 @@ defmodule ExICE.Priv.ConnCheckHandler.Controlled do
         ICEAgent.send_binding_success_response(ice_agent, pair, msg)
 
       %CandidatePair{} = checklist_pair ->
-        checklist_pair = %CandidatePair{checklist_pair | last_seen: pair.last_seen}
-        checklist = Map.put(ice_agent.checklist, checklist_pair.id, checklist_pair)
-        ice_agent = %ICEAgent{ice_agent | checklist: checklist}
+        checklist_pair =
+          if checklist_pair.state == :failed do
+            %CandidatePair{checklist_pair | state: :waiting, last_seen: pair.last_seen}
+          else
+            %CandidatePair{checklist_pair | last_seen: pair.last_seen}
+          end
+
+        ice_agent = put_in(ice_agent.checklist[checklist_pair.id], checklist_pair)
         ICEAgent.send_binding_success_response(ice_agent, checklist_pair, msg)
     end
   end
@@ -42,37 +47,51 @@ defmodule ExICE.Priv.ConnCheckHandler.Controlled do
         ice_agent = %ICEAgent{ice_agent | checklist: checklist}
         ICEAgent.send_binding_success_response(ice_agent, pair, msg)
 
-      %CandidatePair{} = checklist_pair ->
-        if checklist_pair.state == :succeeded do
-          discovered_pair = Map.fetch!(ice_agent.checklist, checklist_pair.discovered_pair_id)
-          discovered_pair = %CandidatePair{discovered_pair | last_seen: pair.last_seen}
-          ice_agent = put_in(ice_agent.checklist[discovered_pair.id], discovered_pair)
-
-          if ice_agent.selected_pair_id == nil do
-            Logger.debug("Nomination request on pair: #{discovered_pair.id}.")
-            update_nominated_flag(ice_agent, discovered_pair.id, true)
-          else
-            ice_agent
-          end
-          |> ICEAgent.send_binding_success_response(discovered_pair, msg)
+      %CandidatePair{state: :succeeded} = checklist_pair ->
+        discovered_pair = Map.fetch!(ice_agent.checklist, checklist_pair.discovered_pair_id)
+        discovered_pair = %CandidatePair{discovered_pair | last_seen: pair.last_seen}
+        ice_agent = put_in(ice_agent.checklist[discovered_pair.id], discovered_pair)
+
+        if ice_agent.selected_pair_id == nil do
+          Logger.debug("Nomination request on pair: #{discovered_pair.id}.")
+          update_nominated_flag(ice_agent, discovered_pair.id, true)
         else
-          # TODO should we check if this pair is not in failed?
-          Logger.debug("""
-          Nomination request on pair that hasn't been verified yet.
-          We will nominate pair once conn check passes.
-          Pair: #{inspect(checklist_pair.id)}
-          """)
+          ice_agent
+        end
+        |> ICEAgent.send_binding_success_response(discovered_pair, msg)
 
-          checklist_pair = %CandidatePair{
-            checklist_pair
-            | nominate?: true,
-              last_seen: pair.last_seen
-          }
+      %CandidatePair{state: :failed} = checklist_pair ->
+        Logger.debug("""
+        Nomination request on failed pair. Re-scheduling pair for conn-check.
+        We will nominate pair once conn check passes.
+        Pair: #{inspect(checklist_pair.id)}
+        """)
 
-          checklist = Map.put(ice_agent.checklist, checklist_pair.id, checklist_pair)
-          ice_agent = %ICEAgent{ice_agent | checklist: checklist}
-          ICEAgent.send_binding_success_response(ice_agent, checklist_pair, msg)
-        end
+        checklist_pair = %CandidatePair{
+          checklist_pair
+          | nominate?: true,
+            last_seen: pair.last_seen,
+            state: :waiting
+        }
+
+        ice_agent = put_in(ice_agent.checklist[checklist_pair.id], checklist_pair)
+        ICEAgent.send_binding_success_response(ice_agent, checklist_pair, msg)
+
+      %CandidatePair{} = checklist_pair ->
+        Logger.debug("""
+        Nomination request on pair that hasn't been verified yet.
+        We will nominate pair once conn check passes.
+        Pair: #{inspect(checklist_pair.id)}
+        """)
+
+        checklist_pair = %CandidatePair{
+          checklist_pair
+          | nominate?: true,
+            last_seen: pair.last_seen
+        }
+
+        ice_agent = put_in(ice_agent.checklist[checklist_pair.id], checklist_pair)
+        ICEAgent.send_binding_success_response(ice_agent, checklist_pair, msg)
     end
   end
 

lib/ex_ice/priv/conn_check_handler/controlling.ex

@@ -34,9 +34,14 @@ defmodule ExICE.Priv.ConnCheckHandler.Controlling do
         ICEAgent.send_binding_success_response(ice_agent, pair, msg)
 
       %CandidatePair{} = checklist_pair ->
-        checklist_pair = %CandidatePair{checklist_pair | last_seen: pair.last_seen}
-        checklist = Map.put(ice_agent.checklist, checklist_pair.id, checklist_pair)
-        ice_agent = %ICEAgent{ice_agent | checklist: checklist}
+        checklist_pair =
+          if checklist_pair.state == :failed do
+            %CandidatePair{checklist_pair | state: :waiting, last_seen: pair.last_seen}
+          else
+            %CandidatePair{checklist_pair | last_seen: pair.last_seen}
+          end
+
+        ice_agent = put_in(ice_agent.checklist[checklist_pair.id], checklist_pair)
         ICEAgent.send_binding_success_response(ice_agent, checklist_pair, msg)
     end
   end

lib/ex_ice/priv/ice_agent.ex

@@ -600,20 +600,7 @@ defmodule ExICE.Priv.ICEAgent do
         %CandidatePair{} =
           pair = Checklist.find_pair(ice_agent.checklist, local_cand.base.id, remote_cand.id)
 
-        case pair.state do
-          :succeeded ->
-            pair = Map.fetch!(ice_agent.checklist, pair.discovered_pair_id)
-            handle_data_message(ice_agent, pair, packet)
-
-          :failed ->
-            Logger.error("Received data on failed pair! Ignoring. Pair id: #{pair.id}")
-            ice_agent
-
-          _other ->
-            # We might receive data on a pair that we haven't checked
-            # on our side yet.
-            handle_data_message(ice_agent, pair, packet)
-        end
+        handle_data_message(ice_agent, pair, packet)
     end
   end
 
@@ -972,7 +959,31 @@ defmodule ExICE.Priv.ICEAgent do
     end
   end
 
+  defp handle_data_message(ice_agent, %{state: :succeeded} = pair, packet) do
+    # take final pair as local candidate might be srflx
+    pair = Map.fetch!(ice_agent.checklist, pair.discovered_pair_id)
+    do_handle_data_message(ice_agent, pair, packet)
+  end
+
+  defp handle_data_message(ice_agent, %{state: :failed} = pair, packet) do
+    Logger.debug("""
+    Received data on failed pair. Rescheduling pair for conn check. Pair id: #{pair.id}\
+    """)
+
+    ice_agent = do_handle_data_message(ice_agent, pair, packet)
+
+    # re-schedule pair
+    pair = %{pair | state: :waiting}
+    ice_agent = put_in(ice_agent.checklist[pair.id], pair)
+    update_ta_timer(ice_agent)
+  end
+
+  # We might receive data on a pair that we haven't check on our side yet.
   defp handle_data_message(ice_agent, pair, packet) do
+    do_handle_data_message(ice_agent, pair, packet)
+  end
+
+  defp do_handle_data_message(ice_agent, pair, packet) do
     pair = %CandidatePair{pair | last_seen: now()}
     ice_agent = put_in(ice_agent.checklist[pair.id], pair)
 
@@ -1058,7 +1069,7 @@ defmodule ExICE.Priv.ICEAgent do
         on: #{inspect({local_cand.base.base_address, local_cand.base.base_port})} \
         """)
 
-        {pair_id, ice_agent} = pop_in(ice_agent.keepalives, msg.transaction_id)
+        {pair_id, ice_agent} = pop_in(ice_agent.keepalives[msg.transaction_id])
 
         pair = Map.fetch!(ice_agent.checklist, pair_id)
         pair = %CandidatePair{pair | last_seen: now()}
@@ -1135,6 +1146,9 @@ defmodule ExICE.Priv.ICEAgent do
         msg,
         use_cand_attr
       )
+      # As a result of handling incoming binding request, we might have re-scheduled pair.
+      # Hence, we have to update ta timer.
+      |> update_ta_timer()
     else
       {:error, reason}
       when reason in [

test/priv/ice_agent_test.exs

@@ -114,7 +114,7 @@ defmodule ExICE.Priv.ICEAgentTest do
     end
   end
 
-  test "don't add pairs with srflx local candidate to the checklist" do
+  test "doesn't add pairs with srflx local candidate to the checklist" do
     ice_agent =
       ICEAgent.new(
         controlling_process: self(),
@@ -150,6 +150,123 @@ defmodule ExICE.Priv.ICEAgentTest do
     assert pair.local_cand_id == host_cand.base.id
   end
 
+  test "forwards data received on a faild pair and re-schedules" do
+    remote_cand = ExICE.Candidate.new(:host, address: {192, 168, 0, 3}, port: 8445)
+
+    ice_agent =
+      ICEAgent.new(
+        controlling_process: self(),
+        role: :controlling,
+        transport_module: Transport.Mock,
+        if_discovery_module: IfDiscovery.Mock
+      )
+      |> ICEAgent.set_remote_credentials("someufrag", "somepwd")
+      |> ICEAgent.gather_candidates()
+      |> ICEAgent.add_remote_candidate(remote_cand)
+
+    [socket] = ice_agent.sockets
+
+    # mark pair as failed
+    [pair] = Map.values(ice_agent.checklist)
+    ice_agent = put_in(ice_agent.checklist[pair.id], %{pair | state: :failed})
+
+    # clear ta_timer, ignore outgoing binding request that has been generated
+    ice_agent = ICEAgent.handle_ta_timeout(ice_agent)
+    assert ice_agent.ta_timer == nil
+
+    # feed some data
+    ice_agent =
+      ICEAgent.handle_udp(ice_agent, socket, remote_cand.address, remote_cand.port, "some data")
+
+    # assert that data has been passed
+    assert_receive {:ex_ice, _pid, {:data, "some data"}}
+
+    # assert that pair is re-scheduled
+    assert [pair] = Map.values(ice_agent.checklist)
+    assert pair.state == :waiting
+    assert ice_agent.ta_timer != nil
+  end
+
+  describe "re-schedules failed pair on incoming binding request" do
+    test "with controlling ice agent" do
+      remote_cand = ExICE.Candidate.new(:host, address: {192, 168, 0, 3}, port: 8445)
+
+      ice_agent =
+        ICEAgent.new(
+          controlling_process: self(),
+          role: :controlling,
+          transport_module: Transport.Mock,
+          if_discovery_module: IfDiscovery.Mock
+        )
+        |> ICEAgent.set_remote_credentials("someufrag", "somepwd")
+        |> ICEAgent.gather_candidates()
+        |> ICEAgent.add_remote_candidate(remote_cand)
+
+      test_rescheduling(ice_agent, remote_cand)
+    end
+
+    test "with controlled ice agent" do
+      remote_cand = ExICE.Candidate.new(:host, address: {192, 168, 0, 3}, port: 8445)
+
+      ice_agent =
+        ICEAgent.new(
+          controlling_process: self(),
+          role: :controlled,
+          transport_module: Transport.Mock,
+          if_discovery_module: IfDiscovery.Mock
+        )
+        |> ICEAgent.set_remote_credentials("someufrag", "somepwd")
+        |> ICEAgent.gather_candidates()
+        |> ICEAgent.add_remote_candidate(remote_cand)
+
+      test_rescheduling(ice_agent, remote_cand)
+    end
+
+    defp test_rescheduling(ice_agent, remote_cand) do
+      [socket] = ice_agent.sockets
+
+      # make sure we won't overflow when modifying tiebreakers later on
+      ice_agent = %{ice_agent | tiebreaker: 100}
+
+      # mark pair as failed
+      [pair] = Map.values(ice_agent.checklist)
+      ice_agent = put_in(ice_agent.checklist[pair.id], %{pair | state: :failed})
+
+      # clear ta_timer, ignore outgoing binding request that has been generated
+      ice_agent = ICEAgent.handle_ta_timeout(ice_agent)
+      assert ice_agent.ta_timer == nil
+
+      # feed incoming binding request
+      ice_attrs =
+        if ice_agent.role == :controlled do
+          [%ICEControlling{tiebreaker: ice_agent.tiebreaker + 1}, %UseCandidate{}]
+        else
+          [%ICEControlled{tiebreaker: ice_agent.tiebreaker - 1}]
+        end
+
+      attrs =
+        [
+          %Username{value: "#{ice_agent.local_ufrag}:someufrag"},
+          %Priority{priority: 1234}
+        ] ++ ice_attrs
+
+      request =
+        Message.new(%Type{class: :request, method: :binding}, attrs)
+        |> Message.with_integrity(ice_agent.local_pwd)
+        |> Message.with_fingerprint()
+
+      raw_request = Message.encode(request)
+
+      ice_agent =
+        ICEAgent.handle_udp(ice_agent, socket, remote_cand.address, remote_cand.port, raw_request)
+
+      # assert that pair is re-scheduled
+      assert [pair] = Map.values(ice_agent.checklist)
+      assert pair.state == :waiting
+      assert ice_agent.ta_timer != nil
+    end
+  end
+
   describe "sends keepalives" do
     setup do
       remote_cand = ExICE.Candidate.new(:host, address: {192, 168, 0, 2}, port: 8445)