Merge pull request #206 from weiss/xep-0198

XEP-0198: Terminate session if stanza queue becomes too large

doc/guide.tex

@@ -983,10 +983,10 @@ This is a detailed description of each option allowed by the listening modules:
   \titem{max\_ack\_queue: Size}
     This option specifies the maximum number of unacknowledged stanzas
     queued for possible retransmission if \term{stream\_management} is
-    enabled. When the limit is reached, the first stanza is dropped from
+    enabled. When the limit is exceeded, the client session is
-    the queue before adding the next one. This option can be specified
+    terminated. This option can be specified for \term{ejabberd\_c2s}
-    for \term{ejabberd\_c2s} listeners. The allowed values are positive
+    listeners. The allowed values are positive integers and
-    integers and \term{infinity}. Default value: \term{500}.
+    \term{infinity}. Default value: \term{500}.
   \titem{max\_fsm\_queue: Size}
     This option specifies the maximum number of elements in the queue of the FSM
     (Finite State Machine).

src/ejabberd_c2s.erl

@@ -1284,6 +1284,10 @@ wait_for_resume(Event, StateData) ->
 %%          {next_state, NextStateName, NextStateData, Timeout} |
 %%          {stop, Reason, NewStateData}
 %%----------------------------------------------------------------------
+handle_event({abort, Xmlelement}, _StateName, StateData) ->
+    send_element(StateData, Xmlelement),
+    send_trailer(StateData),
+    {stop, normal, StateData};
 handle_event(_Event, StateName, StateData) ->
     fsm_next_state(StateName, StateData).
 
@@ -2779,27 +2783,30 @@ mgmt_queue_add(StateData, El) ->
 	       Num ->
 		   Num + 1
 	     end,
-    NewState = limit_queue_length(StateData),
+    NewQueue = queue:in({NewNum, El}, StateData#state.mgmt_queue),
-    NewQueue = queue:in({NewNum, El}, NewState#state.mgmt_queue),
+    NewState = StateData#state{mgmt_queue = NewQueue,
-    NewState#state{mgmt_queue = NewQueue, mgmt_stanzas_out = NewNum}.
+			       mgmt_stanzas_out = NewNum},
+    check_queue_length(NewState).
 
 mgmt_queue_drop(StateData, NumHandled) ->
     NewQueue = jlib:queue_drop_while(fun({N, _Stanza}) -> N =< NumHandled end,
 				     StateData#state.mgmt_queue),
     StateData#state{mgmt_queue = NewQueue}.
 
-limit_queue_length(#state{mgmt_max_queue = Limit} = StateData)
+check_queue_length(#state{mgmt_max_queue = Limit} = StateData)
     when Limit == infinity;
 	 Limit == unlimited ->
     StateData;
-limit_queue_length(#state{jid = JID,
+check_queue_length(#state{mgmt_queue = Queue,
-			  mgmt_queue = Queue,
 			  mgmt_max_queue = Limit} = StateData) ->
-    case queue:len(Queue) >= Limit of
+    case queue:len(Queue) > Limit of
       true ->
-	  ?WARNING_MSG("Dropping stanza from too long ACK queue for ~s",
+	  ?WARNING_MSG("ACK queue too long, terminating session for ~s",
-		       [jlib:jid_to_string(JID)]),
+		       [jlib:jid_to_string(StateData#state.jid)]),
-	  limit_queue_length(StateData#state{mgmt_queue = queue:drop(Queue)});
+	  Lang = StateData#state.lang,
+	  Err = ?SERRT_POLICY_VIOLATION(Lang, <<"Too many unacked stanzas">>),
+	  (?GEN_FSM):send_all_state_event(self(), {abort, Err}),
+	  StateData#state{mgmt_resend = false}; % Don't resend the flood!
       false ->
 	  StateData
     end.