properly track GTPv2 commands

Declaring GTPv2 command as finished prematurly can lead to
duplicated trigger modify bearer commands.
Add a test case for this and fix the message tracking.

src/pgw_s5s8.erl

@@ -257,14 +257,12 @@ handle_request(#request{gtp_port = GtpPort, ip = SrcIP, port = SrcPort} = ReqKey
 			        #v2_bearer_context{
 				   group = #{?'EPS Bearer ID' := EBI} = Bearer}}},
 	       _Resent, #{context := Context} = State) ->
-    gtp_context:request_finished(ReqKey),
-
     RequestIEs0 = [AMBR,
 		   #v2_bearer_context{
 		      group = copy_ies_to_response(Bearer, [EBI], [?'Bearer Level QoS'])}],
     RequestIEs = gtp_v2_c:build_recovery(Context, false, RequestIEs0),
     Msg = msg(Context, update_bearer_request, RequestIEs),
-    send_request(GtpPort, SrcIP, SrcPort, ?T3, ?N3, Msg#gtp{seq_no = SeqNo}, undefined),
+    send_request(GtpPort, SrcIP, SrcPort, ?T3, ?N3, Msg#gtp{seq_no = SeqNo}, ReqKey),
 
     {noreply, State};
 
@@ -330,14 +328,15 @@ handle_request(ReqKey, _Msg, _Resent, State) ->
 handle_response(ReqInfo, #gtp{version = v1} = Msg, Request, State) ->
     ?GTP_v1_Interface:handle_response(ReqInfo, Msg, Request, State);
 
-handle_response(_,
+handle_response(CommandReqKey,
 		#gtp{type = update_bearer_response,
 		     ie = #{?'Cause' := #v2_cause{v2_cause = Cause},
 			    ?'Bearer Contexts to be modified' :=
 			        #v2_bearer_context{
 				   group = #{?'Cause' := #v2_cause{v2_cause = BearerCause}}
 				  }}} = Response,
 		_Request, #{context := Context0} = State) ->
+    gtp_context:request_finished(CommandReqKey),
     Context = gtp_path:bind(Response, Context0),
 
     if Cause =:= request_accepted andalso BearerCause =:= request_accepted ->

src/pgw_s5s8_proxy.erl

@@ -299,8 +299,6 @@ handle_request(ReqKey,
     forward_request(sgw2pgw, ReqKey, Request, State1),
 
     State = trigger_request(sgw2pgw, ReqKey, Request, State1),
-
-    gtp_context:request_finished(ReqKey),
     {noreply, State};
 
 %%
@@ -453,6 +451,8 @@ handle_response(#proxy_request{direction = pgw2sgw} = ProxyRequest,
     lager:warning("OK Response ~p", [lager:pr(Response, ?MODULE)]),
 
     forward_response(ProxyRequest, Response, ProxyContext),
+    trigger_request_finished(Response, State),
+
     {noreply, State};
 
 handle_response(#proxy_request{direction = sgw2pgw} = ProxyRequest,
@@ -727,7 +727,7 @@ forward_request(Direction, ReqKey,
 		#gtp{seq_no = ReqSeqNo,
 		     ie = #{?'Recovery' := Recovery}} = Request,
 		#{last_trigger_id :=
-		      {ReqSeqNo, LastFwdSeqNo, GtpPort, SrcIP, SrcPort}} = State) ->
+		      {ReqSeqNo, LastFwdSeqNo, GtpPort, SrcIP, SrcPort, _}} = State) ->
 
     Context = forward_context(Direction, State),
     FwdReq = build_context_request(Context, false, LastFwdSeqNo, Request),
@@ -747,11 +747,18 @@ trigger_request(Direction, #request{gtp_port = GtpPort, ip = SrcIP, port = SrcPo
     Context = forward_context(Direction, State),
     case ergw_proxy_lib:get_seq_no(Context, ReqKey, Request) of
 	{ok, FwdSeqNo} ->
-	    State#{last_trigger_id => {FwdSeqNo, SeqNo, GtpPort, SrcIP, SrcPort}};
+	    State#{last_trigger_id => {FwdSeqNo, SeqNo, GtpPort, SrcIP, SrcPort, ReqKey}};
 	_ ->
 	    State
     end.
 
+trigger_request_finished(#gtp{seq_no = SeqNo},
+			 #{last_trigger_id :=
+			       {_, SeqNo, _, _, _, CommandReqKey}}) ->
+    gtp_context:request_finished(CommandReqKey);
+trigger_request_finished(_, _) ->
+    ok.
+
 forward_response(#proxy_request{request = ReqKey, seq_no = SeqNo, new_peer = NewPeer},
 		 Response, Context) ->
     GtpResp = build_context_request(Context, NewPeer, SeqNo, Response),

test/ergw_test_lib.erl

@@ -303,7 +303,9 @@ recv_pdu_msg(Response, At, S, IP, SeqNo, Timeout, Fail) ->
 	    Msg;
 	#gtp{} = Msg
 	  when SeqNo =:= undefined ->
-	    Msg
+	    Msg;
+	#gtp{} = Msg ->
+	    recv_pdu_fail(Fail, {unexpected, Msg})
     end.
 
 recv_pdu_fail(Fail, Why) when is_function(Fail) ->

test/pgw_SUITE.erl

@@ -126,6 +126,7 @@ all() ->
      modify_bearer_request_ra_update,
      modify_bearer_request_tei_update,
      modify_bearer_command,
+     modify_bearer_command_resend,
      modify_bearer_command_timeout,
      modify_bearer_command_congestion,
      change_notification_request_with_tei,
@@ -617,6 +618,39 @@ modify_bearer_command(Config) ->
     meck_validate(Config),
     ok.
 
+%%--------------------------------------------------------------------
+modify_bearer_command_resend() ->
+    [{doc, "Check Modify Bearer Command"}].
+modify_bearer_command_resend(Config) ->
+    %% a resend of a Modify Bearer Command should not
+    %% trigger a second Update Bearer Request
+    Cntl = start_gtpc_server(Config),
+    S = make_gtp_socket(0, Config),
+
+    {GtpC1, _, _} = create_session(S),
+    {GtpC2, Req0} = modify_bearer_command(simple, S, GtpC1),
+
+    Req1 = recv_pdu(S, Req0#gtp.seq_no, ?TIMEOUT, ok),
+    validate_response(modify_bearer_command, simple, Req1, GtpC2),
+
+    %% resend Modify Bearer Command...
+    send_pdu(S, Req0),
+    %% ... should not trigger a second request
+    ?equal(timeout, recv_pdu(S, -1, 100, fun(Why) -> Why end)),
+
+    Response = make_response(Req1, simple, GtpC2),
+    send_pdu(S, Response),
+
+    ?equal({ok, timeout}, recv_pdu(S, Req1#gtp.seq_no, ?TIMEOUT, ok)),
+    ?equal([], outstanding_requests()),
+
+    delete_session(S, GtpC2),
+    stop_gtpc_server(Cntl),
+
+    ok = meck:wait(?HUT, terminate, '_', ?TIMEOUT),
+    meck_validate(Config),
+    ok.
+
 %%--------------------------------------------------------------------
 modify_bearer_command_timeout() ->
     [{doc, "Check Modify Bearer Command"}].

test/pgw_proxy_SUITE.erl

@@ -254,6 +254,7 @@ all_tests() ->
      modify_bearer_request_ra_update,
      modify_bearer_request_tei_update,
      modify_bearer_command,
+     modify_bearer_command_resend,
      modify_bearer_command_timeout,
      modify_bearer_command_congestion,
      update_bearer_request,
@@ -777,6 +778,39 @@ modify_bearer_command(Config) ->
     meck_validate(Config),
     ok.
 
+%%--------------------------------------------------------------------
+modify_bearer_command_resend() ->
+    [{doc, "Check Modify Bearer Command"}].
+modify_bearer_command_resend(Config) ->
+    %% a resend of a Modify Bearer Command should not
+    %% trigger a second Update Bearer Request
+    Cntl = start_gtpc_server(Config),
+    S = make_gtp_socket(0, Config),
+
+    {GtpC1, _, _} = create_session(S),
+    {GtpC2, Req0} = modify_bearer_command(simple, S, GtpC1),
+
+    Req1 = recv_pdu(S, Req0#gtp.seq_no, ?TIMEOUT, ok),
+    validate_response(modify_bearer_command, simple, Req1, GtpC2),
+
+    %% resend Modify Bearer Command...
+    send_pdu(S, Req0),
+    %% ... should not trigger a second request
+    ?equal(timeout, recv_pdu(S, -1, 100, fun(Why) -> Why end)),
+
+    Response = make_response(Req1, simple, GtpC2),
+    send_pdu(S, Response),
+
+    ?equal({ok, timeout}, recv_pdu(S, Req1#gtp.seq_no, ?TIMEOUT, ok)),
+    ?equal([], outstanding_requests()),
+
+    delete_session(S, GtpC2),
+    stop_gtpc_server(Cntl),
+
+    ok = meck:wait(?HUT, terminate, '_', ?TIMEOUT),
+    meck_validate(Config),
+    ok.
+
 %%--------------------------------------------------------------------
 modify_bearer_command_timeout() ->
     [{doc, "Check Modify Bearer Command"}].