diff options
| -rw-r--r-- | Makefile | 2 | ||||
| -rw-r--r-- | rabbitmq-components.mk | 3 | ||||
| -rw-r--r-- | src/dtree.erl | 205 | ||||
| -rw-r--r-- | src/rabbit_amqqueue.erl | 4 | ||||
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 19 | ||||
| -rw-r--r-- | src/rabbit_channel.erl | 123 | ||||
| -rw-r--r-- | src/rabbit_msg_store.erl | 73 | ||||
| -rw-r--r-- | src/rabbit_msg_store_gc.erl | 16 | ||||
| -rw-r--r-- | src/rabbit_policies.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_quorum_queue.erl | 50 | ||||
| -rw-r--r-- | src/unconfirmed_messages.erl | 280 | ||||
| -rw-r--r-- | test/confirms_rejects_SUITE.erl | 168 | ||||
| -rw-r--r-- | test/dead_lettering_SUITE.erl | 33 | ||||
| -rw-r--r-- | test/priority_queue_SUITE.erl | 24 | ||||
| -rw-r--r-- | test/queue_parallel_SUITE.erl | 27 | ||||
| -rw-r--r-- | test/quorum_queue_SUITE.erl | 11 | ||||
| -rw-r--r-- | test/simple_ha_SUITE.erl | 23 |
17 files changed, 739 insertions, 324 deletions
@@ -135,7 +135,7 @@ endef LOCAL_DEPS = sasl mnesia os_mon inets compiler public_key crypto ssl syntax_tools BUILD_DEPS = rabbitmq_cli syslog -DEPS = ranch lager rabbit_common ra sysmon_handler stdout_formatter +DEPS = ranch lager rabbit_common ra sysmon_handler stdout_formatter recon observer_cli TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers amqp_client meck proper dep_syslog = git https://github.com/schlagert/syslog 3.4.5 diff --git a/rabbitmq-components.mk b/rabbitmq-components.mk index 3a8d9a5a4f..0ff2ad0253 100644 --- a/rabbitmq-components.mk +++ b/rabbitmq-components.mk @@ -117,7 +117,8 @@ dep_lager = hex 3.6.10 dep_prometheus = git https://github.com/deadtrickster/prometheus.erl v4.3.0 dep_ra = git https://github.com/rabbitmq/ra.git master dep_ranch = hex 1.7.1 -dep_recon = hex 2.4.0 +dep_recon = hex 2.5.0 +dep_observer_cli = hex 1.5.0 dep_stdout_formatter = hex 0.2.2 dep_sysmon_handler = hex 1.1.0 diff --git a/src/dtree.erl b/src/dtree.erl deleted file mode 100644 index 6f4d102c10..0000000000 --- a/src/dtree.erl +++ /dev/null @@ -1,205 +0,0 @@ -%% The contents of this file are subject to the Mozilla Public License -%% Version 1.1 (the "License"); you may not use this file except in -%% compliance with the License. You may obtain a copy of the License -%% at https://www.mozilla.org/MPL/ -%% -%% Software distributed under the License is distributed on an "AS IS" -%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See -%% the License for the specific language governing rights and -%% limitations under the License. -%% -%% The Original Code is RabbitMQ. -%% -%% The Initial Developer of the Original Code is GoPivotal, Inc. -%% Copyright (c) 2007-2019 Pivotal Software, Inc. All rights reserved. -%% - -%% A dual-index tree. -%% -%% Entries have the following shape: -%% -%% +----+--------------------+---+ -%% | PK | SK1, SK2, ..., SKN | V | -%% +----+--------------------+---+ -%% -%% i.e. a primary key, set of secondary keys, and a value. -%% -%% There can be only one entry per primary key, but secondary keys may -%% appear in multiple entries. -%% -%% The set of secondary keys must be non-empty. Or, to put it another -%% way, entries only exist while their secondary key set is non-empty. - --module(dtree). - --export([empty/0, insert/4, take/3, take/2, take_one/2, take_all/2, drop/2, - is_defined/2, is_empty/1, smallest/1, size/1]). - -%%---------------------------------------------------------------------------- - --export_type([?MODULE/0]). - --opaque ?MODULE() :: {gb_trees:tree(), gb_trees:tree()}. - --type pk() :: any(). --type sk() :: any(). --type val() :: any(). --type kv() :: {pk(), val()}. - -%%---------------------------------------------------------------------------- - --spec empty() -> ?MODULE(). - -empty() -> {gb_trees:empty(), gb_trees:empty()}. - -%% Insert an entry. Fails if there already is an entry with the given -%% primary key. - --spec insert(pk(), [sk()], val(), ?MODULE()) -> ?MODULE(). - -insert(PK, [], V, {P, S}) -> - %% dummy insert to force error if PK exists - _ = gb_trees:insert(PK, {gb_sets:empty(), V}, P), - {P, S}; -insert(PK, SKs, V, {P, S}) -> - {gb_trees:insert(PK, {gb_sets:from_list(SKs), V}, P), - lists:foldl(fun (SK, S0) -> - case gb_trees:lookup(SK, S0) of - {value, PKS} -> PKS1 = gb_sets:insert(PK, PKS), - gb_trees:update(SK, PKS1, S0); - none -> PKS = gb_sets:singleton(PK), - gb_trees:insert(SK, PKS, S0) - end - end, S, SKs)}. - -%% Remove the given secondary key from the entries of the given -%% primary keys, returning the primary-key/value pairs of any entries -%% that were dropped as the result (i.e. due to their secondary key -%% set becoming empty). It is ok for the given primary keys and/or -%% secondary key to not exist. - --spec take([pk()], sk(), ?MODULE()) -> {[kv()], ?MODULE()}. - -take(PKs, SK, {P, S}) -> - case gb_trees:lookup(SK, S) of - none -> {[], {P, S}}; - {value, PKS} -> TakenPKS = gb_sets:from_list(PKs), - PKSInter = gb_sets:intersection(PKS, TakenPKS), - PKSDiff = gb_sets_difference (PKS, PKSInter), - {KVs, P1} = take2(PKSInter, SK, P), - {KVs, {P1, case gb_sets:is_empty(PKSDiff) of - true -> gb_trees:delete(SK, S); - false -> gb_trees:update(SK, PKSDiff, S) - end}} - end. - -%% Remove the given secondary key from all entries, returning the -%% primary-key/value pairs of any entries that were dropped as the -%% result (i.e. due to their secondary key set becoming empty). It is -%% ok for the given secondary key to not exist. - --spec take(sk(), ?MODULE()) -> {[kv()], ?MODULE()}. - -take(SK, {P, S}) -> - case gb_trees:lookup(SK, S) of - none -> {[], {P, S}}; - {value, PKS} -> {KVs, P1} = take2(PKS, SK, P), - {KVs, {P1, gb_trees:delete(SK, S)}} - end. - -%% Drop an entry with the primary key and clears secondary keys for this key, -%% returning a list with a key-value pair as a result. -%% If the primary key does not exist, returns an empty list. - --spec take_one(pk(), ?MODULE()) -> {[{pk(), val()}], ?MODULE()}. - -take_one(PK, {P, S}) -> - case gb_trees:lookup(PK, P) of - {value, {SKS, Value}} -> - P1 = gb_trees:delete(PK, P), - S1 = gb_sets:fold( - fun(SK, Acc) -> - {value, PKS} = gb_trees:lookup(SK, Acc), - PKS1 = gb_sets:delete(PK, PKS), - case gb_sets:is_empty(PKS1) of - true -> gb_trees:delete(SK, Acc); - false -> gb_trees:update(SK, PKS1, Acc) - end - end, S, SKS), - {[{PK, Value}], {P1, S1}}; - none -> {[], {P, S}} - end. - -%% Drop all entries which contain the given secondary key, returning -%% the primary-key/value pairs of these entries. It is ok for the -%% given secondary key to not exist. - --spec take_all(sk(), ?MODULE()) -> {[kv()], ?MODULE()}. - -take_all(SK, {P, S}) -> - case gb_trees:lookup(SK, S) of - none -> {[], {P, S}}; - {value, PKS} -> {KVs, SKS, P1} = take_all2(PKS, P), - {KVs, {P1, prune(SKS, PKS, S)}} - end. - -%% Drop all entries for the given primary key (which does not have to exist). - --spec drop(pk(), ?MODULE()) -> ?MODULE(). - -drop(PK, {P, S}) -> - case gb_trees:lookup(PK, P) of - none -> {P, S}; - {value, {SKS, _V}} -> {gb_trees:delete(PK, P), - prune(SKS, gb_sets:singleton(PK), S)} - end. - --spec is_defined(sk(), ?MODULE()) -> boolean(). - -is_defined(SK, {_P, S}) -> gb_trees:is_defined(SK, S). - --spec is_empty(?MODULE()) -> boolean(). - -is_empty({P, _S}) -> gb_trees:is_empty(P). - --spec smallest(?MODULE()) -> kv(). - -smallest({P, _S}) -> {K, {_SKS, V}} = gb_trees:smallest(P), - {K, V}. - --spec size(?MODULE()) -> non_neg_integer(). - -size({P, _S}) -> gb_trees:size(P). - -%%---------------------------------------------------------------------------- - -take2(PKS, SK, P) -> - gb_sets:fold(fun (PK, {KVs, P0}) -> - {SKS, V} = gb_trees:get(PK, P0), - SKS1 = gb_sets:delete(SK, SKS), - case gb_sets:is_empty(SKS1) of - true -> KVs1 = [{PK, V} | KVs], - {KVs1, gb_trees:delete(PK, P0)}; - false -> {KVs, gb_trees:update(PK, {SKS1, V}, P0)} - end - end, {[], P}, PKS). - -take_all2(PKS, P) -> - gb_sets:fold(fun (PK, {KVs, SKS0, P0}) -> - {SKS, V} = gb_trees:get(PK, P0), - {[{PK, V} | KVs], gb_sets:union(SKS, SKS0), - gb_trees:delete(PK, P0)} - end, {[], gb_sets:empty(), P}, PKS). - -prune(SKS, PKS, S) -> - gb_sets:fold(fun (SK0, S0) -> - PKS1 = gb_trees:get(SK0, S0), - PKS2 = gb_sets_difference(PKS1, PKS), - case gb_sets:is_empty(PKS2) of - true -> gb_trees:delete(SK0, S0); - false -> gb_trees:update(SK0, PKS2, S0) - end - end, S, SKS). - -gb_sets_difference(S1, S2) -> - gb_sets:fold(fun gb_sets:delete_any/2, S1, S2). diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index 9e94dd8f27..85c647ae8c 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -761,7 +761,9 @@ check_dlxrk_arg({Type, _}, _Args) -> {error, {unacceptable_type, Type}}. check_overflow({longstr, Val}, _Args) -> - case lists:member(Val, [<<"drop-head">>, <<"reject-publish">>]) of + case lists:member(Val, [<<"drop-head">>, + <<"reject-publish">>, + <<"reject-publish-dlx">>]) of true -> ok; false -> {error, invalid_overflow} end; diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index b3f89b7ef0..2666847e77 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -82,7 +82,7 @@ %% max length in bytes, if configured max_bytes, %% an action to perform if queue is to be over a limit, - %% can be either drop-head (default) or reject-publish + %% can be either drop-head (default), reject-publish or reject-publish-dlx overflow, %% when policies change, this version helps queue %% determine what previously scheduled/set up state to ignore, @@ -704,12 +704,25 @@ maybe_deliver_or_enqueue(Delivery = #delivery{message = Message}, Delivered, State = #q{overflow = Overflow, backing_queue = BQ, - backing_queue_state = BQS}) -> + backing_queue_state = BQS, + dlx = DLX, + dlx_routing_key = RK}) -> send_mandatory(Delivery), %% must do this before confirms case {will_overflow(Delivery, State), Overflow} of {true, 'reject-publish'} -> %% Drop publish and nack to publisher send_reject_publish(Delivery, Delivered, State); + {true, 'reject-publish-dlx'} -> + %% Publish to DLX + with_dlx( + DLX, + fun (X) -> + QName = qname(State), + rabbit_dead_letter:publish(Message, maxlen, X, RK, QName) + end, + fun () -> ok end), + %% Drop publish and nack to publisher + send_reject_publish(Delivery, Delivered, State); _ -> {IsDuplicate, BQS1} = BQ:is_duplicate(Message, BQS), State1 = State#q{backing_queue_state = BQS1}, @@ -766,6 +779,8 @@ maybe_drop_head(State = #q{max_length = undefined, {false, State}; maybe_drop_head(State = #q{overflow = 'reject-publish'}) -> {false, State}; +maybe_drop_head(State = #q{overflow = 'reject-publish-dlx'}) -> + {false, State}; maybe_drop_head(State = #q{overflow = 'drop-head'}) -> maybe_drop_head(false, State). diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index 19d4449e21..7c07aa07f3 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -155,7 +155,7 @@ confirm_enabled, %% publisher confirm delivery tag sequence publish_seqno, - %% a dtree used to track unconfirmed + %% an unconfirmed_messages data structure used to track unconfirmed %% (to publishers) messages unconfirmed, %% a list of tags for published messages that were @@ -518,7 +518,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost, delivering_queues = sets:new(), confirm_enabled = false, publish_seqno = 1, - unconfirmed = dtree:empty(), + unconfirmed = unconfirmed_messages:new(), rejected = [], confirmed = [], reply_consumer = none, @@ -707,10 +707,15 @@ handle_cast({mandatory_received, _MsgSeqNo}, State) -> handle_cast({reject_publish, MsgSeqNo, _QPid}, State = #ch{unconfirmed = UC}) -> %% It does not matter which queue rejected the message, - %% if any queue rejected it - it should not be confirmed. - {MXs, UC1} = dtree:take_one(MsgSeqNo, UC), + %% if any queue did, it should not be confirmed. + {MaybeRejected, UC1} = unconfirmed_messages:reject_msg(MsgSeqNo, UC), %% NB: don't call noreply/1 since we don't want to send confirms. - noreply_coalesce(record_rejects(MXs, State#ch{unconfirmed = UC1})); + case MaybeRejected of + not_confirmed -> + noreply_coalesce(State#ch{unconfirmed = UC1}); + {rejected, MX} -> + noreply_coalesce(record_rejects([MX], State#ch{unconfirmed = UC1})) + end; handle_cast({confirm, MsgSeqNos, QPid}, State) -> noreply_coalesce(confirm(MsgSeqNos, QPid, State)). @@ -766,9 +771,12 @@ handle_info({ra_event, {Name, _} = From, _} = Evt, eol -> State1 = handle_consuming_queue_down_or_eol(Name, State0), State2 = handle_delivering_queue_down(Name, State1), - %% TODO: this should use dtree:take/3 - {MXs, UC1} = dtree:take(Name, State2#ch.unconfirmed), - State3 = record_confirms(MXs, State1#ch{unconfirmed = UC1}), + {ConfirmMXs, RejectMXs, UC1} = + unconfirmed_messages:forget_ref(Name, State2#ch.unconfirmed), + %% Deleted queue is a special case. + %% Do not nack the "rejected" messages. + State3 = record_confirms(ConfirmMXs ++ RejectMXs, + State2#ch{unconfirmed = UC1}), erase_queue_stats(QName), noreply_coalesce( State3#ch{queue_states = maps:remove(Name, QueueStates), @@ -776,7 +784,7 @@ handle_info({ra_event, {Name, _} = From, _} = Evt, end; _ -> %% the assumption here is that the queue state has been cleaned up and - %% this is a residual ra notification + %% this is a residual Ra notification noreply_coalesce(State0) end; @@ -1818,14 +1826,33 @@ track_delivering_queue(NoAck, QPid, QName, false -> sets:add_element(QRef, DQ) end}. -handle_publishing_queue_down(QPid, Reason, State = #ch{unconfirmed = UC}) +handle_publishing_queue_down(QPid, Reason, + State = #ch{unconfirmed = UC, + queue_names = QNames}) when ?IS_CLASSIC(QPid) -> - case rabbit_misc:is_abnormal_exit(Reason) of - true -> {MXs, UC1} = dtree:take_all(QPid, UC), - record_rejects(MXs, State#ch{unconfirmed = UC1}); - false -> {MXs, UC1} = dtree:take(QPid, UC), - record_confirms(MXs, State#ch{unconfirmed = UC1}) - + case maps:get(QPid, QNames, none) of + %% The queue is unknown, the confirm must have been processed already + none -> State; + _QName -> + case {rabbit_misc:is_abnormal_exit(Reason), Reason} of + {true, _} -> + {RejectMXs, UC1} = + unconfirmed_messages:reject_all_for_queue(QPid, UC), + record_rejects(RejectMXs, State#ch{unconfirmed = UC1}); + {false, normal} -> + {ConfirmMXs, RejectMXs, UC1} = + unconfirmed_messages:forget_ref(QPid, UC), + %% Deleted queue is a special case. + %% Do not nack the "rejected" messages. + record_confirms(ConfirmMXs ++ RejectMXs, + State#ch{unconfirmed = UC1}); + {false, _} -> + {ConfirmMXs, RejectMXs, UC1} = + unconfirmed_messages:forget_ref(QPid, UC), + State1 = record_confirms(ConfirmMXs, + State#ch{unconfirmed = UC1}), + record_rejects(RejectMXs, State1) + end end; handle_publishing_queue_down(QPid, _Reason, _State) when ?IS_QUORUM(QPid) -> error(quorum_queues_should_never_be_monitored). @@ -2095,13 +2122,13 @@ notify_limiter(Limiter, Acked) -> %% common case. case rabbit_limiter:is_active(Limiter) of false -> ok; - true -> case lists:foldl(fun ({_, CTag, _}, Acc) when is_integer(CTag) -> + true -> case lists:foldl(fun ({_, CTag, _, _}, Acc) when is_integer(CTag) -> %% Quorum queues use integer CTags %% classic queues use binaries %% Quorum queues do not interact %% with limiters Acc; - ({_, _, _}, Acc) -> Acc + 1 + ({_, _, _, _}, Acc) -> Acc + 1 end, 0, Acked) of 0 -> ok; Count -> rabbit_limiter:ack(Limiter, Count) @@ -2139,21 +2166,33 @@ deliver_to_queues({Delivery = #delivery{message = Message = #basic_message{ %% frequently would in fact be more expensive in the common case. {QNames1, QMons1} = lists:foldl(fun (Q, {QNames0, QMons0}) when ?is_amqqueue(Q) -> - QPid = amqqueue:get_pid(Q), - QRef = qpid_to_ref(QPid), - QName = amqqueue:get_name(Q), - {case maps:is_key(QRef, QNames0) of - true -> QNames0; - false -> maps:put(QRef, QName, QNames0) - end, maybe_monitor(QPid, QMons0)} - end, {QNames, maybe_monitor_all(DeliveredQPids, QMons)}, Qs), + QPid = amqqueue:get_pid(Q), + QRef = qpid_to_ref(QPid), + QName = amqqueue:get_name(Q), + case ?IS_CLASSIC(QRef) of + true -> + SPids = amqqueue:get_slave_pids(Q), + NewQNames = + maps:from_list([{Ref, QName} || Ref <- [QRef | SPids]]), + {maps:merge(NewQNames, QNames0), + maybe_monitor_all([QPid | SPids], QMons0)}; + false -> + {maps:put(QRef, QName, QNames0), QMons0} + end + end, + {QNames, QMons}, Qs), State1 = State#ch{queue_names = QNames1, queue_monitors = QMons1}, %% NB: the order here is important since basic.returns must be %% sent before confirms. ok = process_routing_mandatory(Mandatory, AllDeliveredQRefs, Message, State1), - State2 = process_routing_confirm(Confirm, AllDeliveredQRefs , MsgSeqNo, + AllDeliveredQNames = [ QName || QRef <- AllDeliveredQRefs, + {ok, QName} <- [maps:find(QRef, QNames1)]], + State2 = process_routing_confirm(Confirm, + AllDeliveredQRefs, + AllDeliveredQNames, + MsgSeqNo, XName, State1), case rabbit_event:stats_level(State, #ch.stats_timer) of fine -> @@ -2179,16 +2218,22 @@ process_routing_mandatory(_Mandatory = false, process_routing_mandatory(_, _, _, _) -> ok. -process_routing_confirm(false, _, _, _, State) -> +process_routing_confirm(false, _, _, _, _, State) -> State; -process_routing_confirm(true, [], MsgSeqNo, XName, State) -> +process_routing_confirm(true, [], _, MsgSeqNo, XName, State) -> record_confirms([{MsgSeqNo, XName}], State); -process_routing_confirm(true, QRefs, MsgSeqNo, XName, State) -> - State#ch{unconfirmed = dtree:insert(MsgSeqNo, QRefs, XName, - State#ch.unconfirmed)}. - -confirm(MsgSeqNos, QRef, State = #ch{unconfirmed = UC}) -> - {MXs, UC1} = dtree:take(MsgSeqNos, QRef, UC), +process_routing_confirm(true, QRefs, QNames, MsgSeqNo, XName, State) -> + State#ch{unconfirmed = + unconfirmed_messages:insert(MsgSeqNo, QNames, QRefs, XName, + State#ch.unconfirmed)}. + +confirm(MsgSeqNos, QRef, State = #ch{queue_names = QNames, unconfirmed = UC}) -> + %% NOTE: if queue name does not exist here it's likely that the ref also + %% does not exist in unconfirmed messages. + %% Neither does the 'ignore' atom, so it's a reasonable fallback. + QName = maps:get(QRef, QNames, ignore), + {MXs, UC1} = + unconfirmed_messages:confirm_multiple_msg_ref(MsgSeqNos, QName, QRef, UC), %% NB: don't call noreply/1 since we don't want to send confirms. record_confirms(MXs, State#ch{unconfirmed = UC1}). @@ -2250,9 +2295,9 @@ send_confirms(Cs, Rs, State) -> coalesce_and_send(MsgSeqNos, NegativeMsgSeqNos, MkMsgFun, State = #ch{unconfirmed = UC}) -> SMsgSeqNos = lists:usort(MsgSeqNos), - UnconfirmedCutoff = case dtree:is_empty(UC) of + UnconfirmedCutoff = case unconfirmed_messages:is_empty(UC) of true -> lists:last(SMsgSeqNos) + 1; - false -> {SeqNo, _XName} = dtree:smallest(UC), SeqNo + false -> unconfirmed_messages:smallest(UC) end, Cutoff = lists:min([UnconfirmedCutoff | NegativeMsgSeqNos]), {Ms, Ss} = lists:splitwith(fun(X) -> X < Cutoff end, SMsgSeqNos), @@ -2271,7 +2316,7 @@ ack_len(Acks) -> lists:sum([length(L) || {ack, L} <- Acks]). maybe_complete_tx(State = #ch{tx = {_, _}}) -> State; maybe_complete_tx(State = #ch{unconfirmed = UC}) -> - case dtree:is_empty(UC) of + case unconfirmed_messages:is_empty(UC) of false -> State; true -> complete_tx(State#ch{confirmed = []}) end. @@ -2311,7 +2356,7 @@ i(confirm, #ch{confirm_enabled = CE}) -> CE; i(source, #ch{cfg = #conf{source = ChSrc}}) -> ChSrc; i(name, State) -> name(State); i(consumer_count, #ch{consumer_mapping = CM}) -> maps:size(CM); -i(messages_unconfirmed, #ch{unconfirmed = UC}) -> dtree:size(UC); +i(messages_unconfirmed, #ch{unconfirmed = UC}) -> unconfirmed_messages:size(UC); i(messages_unacknowledged, #ch{unacked_message_q = UAMQ}) -> ?QUEUE:len(UAMQ); i(messages_uncommitted, #ch{tx = {Msgs, _Acks}}) -> ?QUEUE:len(Msgs); i(messages_uncommitted, #ch{}) -> 0; diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl index 1ec5fe9e1a..e3b23cfbca 100644 --- a/src/rabbit_msg_store.erl +++ b/src/rabbit_msg_store.erl @@ -23,7 +23,7 @@ client_ref/1, close_all_indicated/1, write/3, write_flow/3, read/2, contains/2, remove/2]). --export([set_maximum_since_use/2, has_readers/2, combine_files/3, +-export([set_maximum_since_use/2, combine_files/3, delete_file/2]). %% internal -export([transform_dir/3, force_recovery/2]). %% upgrade @@ -1970,33 +1970,48 @@ cleanup_after_file_deletion(File, %% garbage collection / compaction / aggregation -- external %%---------------------------------------------------------------------------- --spec has_readers(non_neg_integer(), gc_state()) -> boolean(). - -has_readers(File, #gc_state { file_summary_ets = FileSummaryEts }) -> - [#file_summary { locked = true, readers = Count }] = - ets:lookup(FileSummaryEts, File), - Count /= 0. - -spec combine_files(non_neg_integer(), non_neg_integer(), gc_state()) -> - deletion_thunk(). + {ok, deletion_thunk()} | {defer, non_neg_integer()}. combine_files(Source, Destination, - State = #gc_state { file_summary_ets = FileSummaryEts, - file_handles_ets = FileHandlesEts, - dir = Dir, - msg_store = Server }) -> - [#file_summary { + State = #gc_state { file_summary_ets = FileSummaryEts }) -> + [#file_summary{locked = true} = SourceSummary] = + ets:lookup(FileSummaryEts, Source), + + [#file_summary{locked = true} = DestinationSummary] = + ets:lookup(FileSummaryEts, Destination), + + case {SourceSummary, DestinationSummary} of + {#file_summary{readers = 0}, #file_summary{readers = 0}} -> + {ok, do_combine_files(SourceSummary, DestinationSummary, + Source, Destination, State)}; + _ -> + rabbit_log:debug("Asked to combine files ~p and ~p but they have active readers. Deferring.", + [Source, Destination]), + DeferredFiles = [FileSummary#file_summary.file + || FileSummary <- [SourceSummary, DestinationSummary], + FileSummary#file_summary.readers /= 0], + {defer, DeferredFiles} + end. + +do_combine_files(SourceSummary, DestinationSummary, + Source, Destination, + State = #gc_state { file_summary_ets = FileSummaryEts, + file_handles_ets = FileHandlesEts, + dir = Dir, + msg_store = Server }) -> + #file_summary { readers = 0, left = Destination, valid_total_size = SourceValid, file_size = SourceFileSize, - locked = true }] = ets:lookup(FileSummaryEts, Source), - [#file_summary { + locked = true } = SourceSummary, + #file_summary { readers = 0, right = Source, valid_total_size = DestinationValid, file_size = DestinationFileSize, - locked = true }] = ets:lookup(FileSummaryEts, Destination), + locked = true } = DestinationSummary, SourceName = filenum_to_name(Source), DestinationName = filenum_to_name(Destination), @@ -2053,22 +2068,30 @@ combine_files(Source, Destination, {#file_summary.file_size, TotalValidData}]), Reclaimed = SourceFileSize + DestinationFileSize - TotalValidData, + rabbit_log:debug("Combined segment files number ~p (source) and ~p (destination), reclaimed ~p bytes", + [Source, Destination, Reclaimed]), gen_server2:cast(Server, {combine_files, Source, Destination, Reclaimed}), safe_file_delete_fun(Source, Dir, FileHandlesEts). --spec delete_file(non_neg_integer(), gc_state()) -> deletion_thunk(). +-spec delete_file(non_neg_integer(), gc_state()) -> {ok, deletion_thunk()} | {defer, non_neg_integer()}. delete_file(File, State = #gc_state { file_summary_ets = FileSummaryEts, file_handles_ets = FileHandlesEts, dir = Dir, msg_store = Server }) -> - [#file_summary { valid_total_size = 0, - locked = true, - file_size = FileSize, - readers = 0 }] = ets:lookup(FileSummaryEts, File), - {[], 0} = load_and_vacuum_message_file(File, State), - gen_server2:cast(Server, {delete_file, File, FileSize}), - safe_file_delete_fun(File, Dir, FileHandlesEts). + case ets:lookup(FileSummaryEts, File) of + [#file_summary { valid_total_size = 0, + locked = true, + file_size = FileSize, + readers = 0 }] -> + {[], 0} = load_and_vacuum_message_file(File, State), + gen_server2:cast(Server, {delete_file, File, FileSize}), + {ok, safe_file_delete_fun(File, Dir, FileHandlesEts)}; + [#file_summary{readers = Readers}] when Readers > 0 -> + rabbit_log:debug("Asked to delete file ~p but it has active readers. Deferring.", + [File]), + {defer, [File]} + end. load_and_vacuum_message_file(File, State = #gc_state { dir = Dir }) -> %% Messages here will be end-of-file at start-of-list diff --git a/src/rabbit_msg_store_gc.erl b/src/rabbit_msg_store_gc.erl index dfadd5586d..60702b5b95 100644 --- a/src/rabbit_msg_store_gc.erl +++ b/src/rabbit_msg_store_gc.erl @@ -119,15 +119,13 @@ attempt_action(Action, Files, State = #state { pending_no_readers = Pending, on_action = Thunks, msg_store_state = MsgStoreState }) -> - case [File || File <- Files, - rabbit_msg_store:has_readers(File, MsgStoreState)] of - [] -> State #state { - on_action = lists:filter( - fun (Thunk) -> not Thunk() end, - [do_action(Action, Files, MsgStoreState) | - Thunks]) }; - [File | _] -> Pending1 = maps:put(File, {Action, Files}, Pending), - State #state { pending_no_readers = Pending1 } + case do_action(Action, Files, MsgStoreState) of + {ok, OkThunk} -> + State#state{on_action = lists:filter(fun (Thunk) -> not Thunk() end, + [OkThunk | Thunks])}; + {defer, [File | _]} -> + Pending1 = maps:put(File, {Action, Files}, Pending), + State #state { pending_no_readers = Pending1 } end. do_action(combine, [Source, Destination], MsgStoreState) -> diff --git a/src/rabbit_policies.erl b/src/rabbit_policies.erl index 7878bed02d..c4f4226448 100644 --- a/src/rabbit_policies.erl +++ b/src/rabbit_policies.erl @@ -131,6 +131,8 @@ validate_policy0(<<"overflow">>, <<"drop-head">>) -> ok; validate_policy0(<<"overflow">>, <<"reject-publish">>) -> ok; +validate_policy0(<<"overflow">>, <<"reject-publish-dlx">>) -> + ok; validate_policy0(<<"overflow">>, Value) -> {error, "~p is not a valid overflow value", [Value]}; diff --git a/src/rabbit_quorum_queue.erl b/src/rabbit_quorum_queue.erl index 8258584e3c..75bd9515f1 100644 --- a/src/rabbit_quorum_queue.erl +++ b/src/rabbit_quorum_queue.erl @@ -644,7 +644,6 @@ cluster_state(Name) -> end. -spec status(rabbit_types:vhost(), Name :: rabbit_misc:resource_name()) -> rabbit_types:infos() | {error, term()}. - status(Vhost, QueueName) -> %% Handle not found queues QName = #resource{virtual_host = Vhost, name = QueueName, kind = queue}, @@ -653,19 +652,52 @@ status(Vhost, QueueName) -> {ok, Q} when ?amqqueue_is_classic(Q) -> {error, classic_queue_not_supported}; {ok, Q} when ?amqqueue_is_quorum(Q) -> - {_, Leader} = amqqueue:get_pid(Q), Nodes = amqqueue:get_quorum_nodes(Q), - Info = [{leader, Leader}, {members, Nodes}], - case ets:lookup(ra_state, RName) of - [{_, State}] -> - [{local_state, State} | Info]; - [] -> - Info - end; + [begin + case get_sys_status({RName, N}) of + {ok, Sys} -> + {_, M} = lists:keyfind(ra_server_state, 1, Sys), + {_, RaftState} = lists:keyfind(raft_state, 1, Sys), + #{commit_index := Commit, + machine_version := MacVer, + current_term := Term, + log := #{last_index := Last, + snapshot_index := SnapIdx}} = M, + [{<<"Node Name">>, N}, + {<<"Raft State">>, RaftState}, + {<<"Log Index">>, Last}, + {<<"Commit Index">>, Commit}, + {<<"Snapshot Index">>, SnapIdx}, + {<<"Term">>, Term}, + {<<"Machine Version">>, MacVer} + ]; + {error, Err} -> + [{<<"Node Name">>, N}, + {<<"Raft State">>, Err}, + {<<"Log Index">>, <<>>}, + {<<"Commit Index">>, <<>>}, + {<<"Snapshot Index">>, <<>>}, + {<<"Term">>, <<>>}, + {<<"Machine Version">>, <<>>} + ] + end + end || N <- Nodes]; {error, not_found} = E -> E end. +get_sys_status(Proc) -> + try lists:nth(5, element(4, sys:get_status(Proc))) of + Sys -> {ok, Sys} + catch + _:Err when is_tuple(Err) -> + {error, element(1, Err)}; + _:_ -> + {error, other} + + end. + + add_member(VHost, Name, Node) -> QName = #resource{virtual_host = VHost, name = Name, kind = queue}, case rabbit_amqqueue:lookup(QName) of diff --git a/src/unconfirmed_messages.erl b/src/unconfirmed_messages.erl new file mode 100644 index 0000000000..63a504a239 --- /dev/null +++ b/src/unconfirmed_messages.erl @@ -0,0 +1,280 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License +%% at https://www.mozilla.org/MPL/ +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and +%% limitations under the License. +%% +%% The Original Code is RabbitMQ. +%% +%% The Initial Developer of the Original Code is GoPivotal, Inc. +%% Copyright (c) 2007-2019 Pivotal Software, Inc. All rights reserved. +%% + +%% Unconfirmed messages tracking. +%% +%% A message should be confirmed to the publisher only when all queues confirm. +%% +%% Messages are published to multiple queues while each queue may be +%% represented by several processes (queue refs). +%% +%% Queue refs return confirmations, rejections, can fail or disconnect. +%% If a queue ref fails, messgae should be rejected. +%% If all queue refs for a queue disconnect (not fail) without confirmation, +%% messge should be rejected. +%% +%% For simplicity, disconnects do not return a reject until all message refs +%% confirm or disconnect. + +-module(unconfirmed_messages). + +-export([new/0, + insert/5, + confirm_msg_ref/4, + confirm_multiple_msg_ref/4, + forget_ref/2, + + reject_msg/2, + reject_all_for_queue/2, + + smallest/1, + size/1, + is_empty/1]). + +%%---------------------------------------------------------------------------- + +-export_type([?MODULE/0]). +-define(SET_VALUE, []). + +-type queue_ref() :: term(). +-type msg_id() :: term(). +-type queue_name() :: rabbit_amqqueue:name(). +-type exchange_name() :: rabbit_exchange:name(). +-type map_set(Type) :: #{Type => ?SET_VALUE}. + +-record(msg_status, { + %% a set of refs waiting for confirm + refs = #{} :: map_set(queue_ref()), + %% shows which queues had at least one confirmation + queue_status = #{} :: #{queue_name() => confirmed | rejected}, + exchange :: exchange_name() +}). + +-record(unconfirmed, { + %% needed to get unconfirmed cutoff + ordered = gb_sets:new() :: gb_sets:set(msg_id()), + %% contains message statuses of all message IDs + index = #{} :: #{msg_id() => #msg_status{}}, + %% needed to look up message IDs for a queue ref + reverse = #{} :: #{queue_ref() => #{msg_id() => ?SET_VALUE}} +}). + +-opaque ?MODULE() :: #unconfirmed{}. + +%%---------------------------------------------------------------------------- + +-spec new() -> ?MODULE(). +new() -> #unconfirmed{}. + +%% Insert an entry for the message ID. Fails if there already is +%% an entry with the given ID. +-spec insert(msg_id(), [queue_name()], [queue_ref()], exchange_name(), ?MODULE()) -> ?MODULE(). +insert(MsgId, QueueNames, QueueRefs, XName, + #unconfirmed{ordered = Ordered, + index = Index, + reverse = Reverse} = UC) -> + case maps:get(MsgId, Index, none) of + none -> + UC#unconfirmed{ + ordered = gb_sets:add(MsgId, Ordered), + index = + Index#{MsgId => + #msg_status{ + refs = maps:from_list([{QR, ?SET_VALUE} || QR <- QueueRefs]), + queue_status = maps:from_list([{QN, rejected} || QN <- QueueNames]), + exchange = XName}}, + reverse = lists:foldl( + fun + (Ref, R) -> + case R of + #{Ref := MsgIdsSet} -> + R#{Ref => MsgIdsSet#{MsgId => ?SET_VALUE}}; + _ -> + R#{Ref => #{MsgId => ?SET_VALUE}} + end + end, + Reverse, QueueRefs) + }; + _ -> + error({message_already_exists, MsgId, QueueNames, QueueRefs, XName, UC}) + end. + +%% Confirms a message on behalf of the given queue. If it was the last queue (ref) +%% on the waiting list, returns 'confirmed' and performs the necessary cleanup. +-spec confirm_msg_ref(msg_id(), queue_name(), queue_ref(), ?MODULE()) -> + {{confirmed | rejected, {msg_id(), exchange_name()}} | not_confirmed, ?MODULE()}. +confirm_msg_ref(MsgId, QueueName, QueueRef, + #unconfirmed{reverse = Reverse} = UC) -> + remove_msg_ref(confirm, MsgId, QueueName, QueueRef, + UC#unconfirmed{reverse = remove_from_reverse(QueueRef, [MsgId], Reverse)}). + +-spec confirm_multiple_msg_ref(msg_id(), queue_name(), queue_ref(), ?MODULE()) -> + {{confirmed | rejected, {msg_id(), exchange_name()}} | not_confirmed, ?MODULE()}. +confirm_multiple_msg_ref(MsgIds, QueueName, QueueRef, + #unconfirmed{reverse = Reverse} = UC0) -> + lists:foldl( + fun(MsgId, {C, UC}) -> + case remove_msg_ref(confirm, MsgId, QueueName, QueueRef, UC) of + {{confirmed, V}, UC1} -> {[V | C], UC1}; + {not_confirmed, UC1} -> {C, UC1} + end + end, + {[], UC0#unconfirmed{reverse = remove_from_reverse(QueueRef, MsgIds, Reverse)}}, + MsgIds). + +%% Removes all messages for a queue. +%% Returns lists of confirmed and rejected messages. +%% +%% If there are no more refs left for the message, either +%% 'confirmed' or 'rejected'. +%% 'confirmed' is returned if all queues have confirmed the message. +-spec forget_ref(queue_ref(), ?MODULE()) -> + {Confirmed :: [{msg_id(), exchange_name()}], + Rejected :: [{msg_id(), exchange_name()}], + ?MODULE()}. +forget_ref(QueueRef, #unconfirmed{reverse = Reverse0} = UC0) -> + MsgIds = maps:keys(maps:get(QueueRef, Reverse0, #{})), + lists:foldl(fun(MsgId, {C, R, UC}) -> + case remove_msg_ref(no_confirm, MsgId, ignore, QueueRef, UC) of + {not_confirmed, UC1} -> {C, R, UC1}; + {{confirmed, V}, UC1} -> {[V | C], R, UC1}; + {{rejected, V}, UC1} -> {C, [V | R], UC1} + end + end, + {[], [], UC0#unconfirmed{reverse = maps:remove(QueueRef, Reverse0)}}, + MsgIds). + +%% Rejects a single message with the given ID. +%% Returns 'rejected' if there was a message with +%% such ID. +-spec reject_msg(msg_id(), ?MODULE()) -> + {{rejected, {msg_id(), exchange_name()}} | not_confirmed, ?MODULE()}. +reject_msg(MsgId, #unconfirmed{ordered = Ordered, index = Index, reverse = Reverse} = UC) -> + case maps:get(MsgId, Index, none) of + none -> + {not_confirmed, UC}; + #msg_status{exchange = XName, + refs = Refs} -> + {{rejected, {MsgId, XName}}, + UC#unconfirmed{ordered = gb_sets:del_element(MsgId, Ordered), + index = maps:remove(MsgId, Index), + reverse = remove_multiple_from_reverse(maps:keys(Refs), [MsgId], Reverse)}} + end. + +%% Rejects all pending messages for a queue. +-spec reject_all_for_queue(queue_ref(), ?MODULE()) -> + {Rejected :: [{msg_id(), exchange_name()}], ?MODULE()}. +reject_all_for_queue(QueueRef, #unconfirmed{reverse = Reverse0} = UC0) -> + MsgIds = maps:keys(maps:get(QueueRef, Reverse0, #{})), + lists:foldl(fun(MsgId, {R, UC}) -> + case reject_msg(MsgId, UC) of + {not_confirmed, UC1} -> {R, UC1}; + {{rejected, V}, UC1} -> {[V | R], UC1} + end + end, + {[], UC0#unconfirmed{reverse = maps:remove(QueueRef, Reverse0)}}, + MsgIds). + +%% Returns a smallest message id. +-spec smallest(?MODULE()) -> msg_id(). +smallest(#unconfirmed{ordered = Ordered}) -> + gb_sets:smallest(Ordered). + +-spec size(?MODULE()) -> msg_id(). +size(#unconfirmed{index = Index}) -> maps:size(Index). + +-spec is_empty(?MODULE()) -> boolean(). +is_empty(#unconfirmed{index = Index, reverse = Reverse, ordered = Ordered} = UC) -> + case maps:size(Index) == 0 of + true -> + %% Assertion + case maps:size(Reverse) == gb_sets:size(Ordered) + andalso + maps:size(Reverse) == 0 of + true -> ok; + false -> error({size_mismatch, UC}) + end, + true; + _ -> + false + end. + +-spec remove_from_reverse(queue_ref(), [msg_id()], + #{queue_ref() => #{msg_id() => ?SET_VALUE}}) -> + #{queue_ref() => #{msg_id() => ?SET_VALUE}}. +remove_from_reverse(QueueRef, MsgIds, Reverse) when is_list(MsgIds) -> + case maps:get(QueueRef, Reverse, none) of + none -> + Reverse; + MsgIdsSet -> + NewMsgIdsSet = maps:without(MsgIds, MsgIdsSet), + case maps:size(NewMsgIdsSet) > 0 of + true -> Reverse#{QueueRef => NewMsgIdsSet}; + false -> maps:remove(QueueRef, Reverse) + end + end. + +-spec remove_multiple_from_reverse([queue_ref()], [msg_id()], + #{queue_ref() => #{msg_id() => ?SET_VALUE}}) -> + #{queue_ref() => #{msg_id() => ?SET_VALUE}}. +remove_multiple_from_reverse(Refs, MsgIds, Reverse0) -> + lists:foldl( + fun(Ref, Reverse) -> + remove_from_reverse(Ref, MsgIds, Reverse) + end, + Reverse0, + Refs). + +-spec remove_msg_ref(confirm | no_confirm, msg_id(), queue_name(), queue_ref(), ?MODULE()) -> + {{confirmed | rejected, {msg_id(), exchange_name()}} | not_confirmed, + ?MODULE()}. +remove_msg_ref(Confirm, MsgId, QueueName, QueueRef, + #unconfirmed{ordered = Ordered, index = Index} = UC) -> + case maps:get(MsgId, Index, none) of + none -> + {not_confirmed, UC}; + #msg_status{refs = #{QueueRef := ?SET_VALUE} = Refs, + queue_status = QStatus, + exchange = XName} = MsgStatus -> + QStatus1 = case {Confirm, QueueName} of + {no_confirm, _} -> QStatus; + {_, ignore} -> QStatus; + {confirm, _} -> QStatus#{QueueName => confirmed} + end, + case maps:size(Refs) == 1 of + true -> + {{confirm_status(QStatus1), {MsgId, XName}}, + UC#unconfirmed{ + ordered = gb_sets:del_element(MsgId, Ordered), + index = maps:remove(MsgId, Index)}}; + false -> + {not_confirmed, + UC#unconfirmed{ + index = Index#{MsgId => + MsgStatus#msg_status{ + refs = maps:remove(QueueRef, Refs), + queue_status = QStatus1}}}} + end; + _ -> {not_confirmed, UC} + end. + +-spec confirm_status(#{queue_name() => confirmed | rejected}) -> confirmed | rejected. +confirm_status(QueueStatus) -> + case lists:all(fun(confirmed) -> true; (_) -> false end, + maps:values(QueueStatus)) of + true -> confirmed; + false -> rejected + end. diff --git a/test/confirms_rejects_SUITE.erl b/test/confirms_rejects_SUITE.erl index 722b5240fc..402bca8737 100644 --- a/test/confirms_rejects_SUITE.erl +++ b/test/confirms_rejects_SUITE.erl @@ -11,11 +11,17 @@ all() -> ]. groups() -> + OverflowTests = [ + confirms_rejects_conflict, + policy_resets_to_default + ], [ {parallel_tests, [parallel], [ - confirms_rejects_conflict, - policy_resets_to_default - ]} + {overflow_reject_publish_dlx, [parallel], OverflowTests}, + {overflow_reject_publish, [parallel], OverflowTests}, + dead_queue_rejects, + mixed_dead_alive_queues_reject + ]} ]. init_per_suite(Config) -> @@ -26,6 +32,14 @@ end_per_suite(Config) -> rabbit_ct_helpers:run_teardown_steps(Config). +init_per_group(overflow_reject_publish, Config) -> + rabbit_ct_helpers:set_config(Config, [ + {overflow, <<"reject-publish">>} + ]); +init_per_group(overflow_reject_publish_dlx, Config) -> + rabbit_ct_helpers:set_config(Config, [ + {overflow, <<"reject-publish-dlx">>} + ]); init_per_group(Group, Config) -> ClusterSize = 2, Config1 = rabbit_ct_helpers:set_config(Config, [ @@ -36,6 +50,10 @@ init_per_group(Group, Config) -> rabbit_ct_broker_helpers:setup_steps() ++ rabbit_ct_client_helpers:setup_steps()). +end_per_group(overflow_reject_publish, _Config) -> + ok; +end_per_group(overflow_reject_publish_dlx, _Config) -> + ok; end_per_group(_Group, Config) -> rabbit_ct_helpers:run_steps(Config, rabbit_ct_client_helpers:teardown_steps() ++ @@ -45,7 +63,10 @@ init_per_testcase(policy_resets_to_default = Testcase, Config) -> Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config), rabbit_ct_helpers:testcase_started( rabbit_ct_helpers:set_config(Config, [{conn, Conn}]), Testcase); -init_per_testcase(confirms_rejects_conflict = Testcase, Config) -> +init_per_testcase(Testcase, Config) + when Testcase == confirms_rejects_conflict; + Testcase == dead_queue_rejects; + Testcase == mixed_dead_alive_queues_reject -> Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config), Conn1 = rabbit_ct_client_helpers:open_unmanaged_connection(Config), @@ -55,7 +76,9 @@ init_per_testcase(confirms_rejects_conflict = Testcase, Config) -> end_per_testcase(policy_resets_to_default = Testcase, Config) -> {_, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0), - amqp_channel:call(Ch, #'queue.delete'{queue = <<"policy_resets_to_default">>}), + XOverflow = ?config(overflow, Config), + QueueName = <<"policy_resets_to_default", "_", XOverflow/binary>>, + amqp_channel:call(Ch, #'queue.delete'{queue = QueueName}), rabbit_ct_client_helpers:close_channels_and_connection(Config, 0), Conn = ?config(conn, Config), @@ -65,7 +88,22 @@ end_per_testcase(policy_resets_to_default = Testcase, Config) -> rabbit_ct_helpers:testcase_finished(Config, Testcase); end_per_testcase(confirms_rejects_conflict = Testcase, Config) -> {_, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0), - amqp_channel:call(Ch, #'queue.delete'{queue = <<"confirms_rejects_conflict">>}), + XOverflow = ?config(overflow, Config), + QueueName = <<"confirms_rejects_conflict", "_", XOverflow/binary>>, + amqp_channel:call(Ch, #'queue.delete'{queue = QueueName}), + end_per_testcase0(Testcase, Config); +end_per_testcase(dead_queue_rejects = Testcase, Config) -> + {_, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0), + amqp_channel:call(Ch, #'queue.delete'{queue = <<"dead_queue_rejects">>}), + end_per_testcase0(Testcase, Config); +end_per_testcase(mixed_dead_alive_queues_reject = Testcase, Config) -> + {_, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0), + amqp_channel:call(Ch, #'queue.delete'{queue = <<"mixed_dead_alive_queues_reject_dead">>}), + amqp_channel:call(Ch, #'queue.delete'{queue = <<"mixed_dead_alive_queues_reject_alive">>}), + amqp_channel:call(Ch, #'exchange.delete'{exchange = <<"mixed_dead_alive_queues_reject">>}), + end_per_testcase0(Testcase, Config). + +end_per_testcase0(Testcase, Config) -> rabbit_ct_client_helpers:close_channels_and_connection(Config, 0), Conn = ?config(conn, Config), @@ -74,9 +112,90 @@ end_per_testcase(confirms_rejects_conflict = Testcase, Config) -> rabbit_ct_client_helpers:close_connection(Conn), rabbit_ct_client_helpers:close_connection(Conn1), + clean_acks_mailbox(), + rabbit_ct_helpers:testcase_finished(Config, Testcase). +dead_queue_rejects(Config) -> + Conn = ?config(conn, Config), + {ok, Ch} = amqp_connection:open_channel(Conn), + QueueName = <<"dead_queue_rejects">>, + amqp_channel:call(Ch, #'confirm.select'{}), + amqp_channel:register_confirm_handler(Ch, self()), + + amqp_channel:call(Ch, #'queue.declare'{queue = QueueName, + durable = true}), + + amqp_channel:call(Ch, #'basic.publish'{routing_key = QueueName}, + #amqp_msg{payload = <<"HI">>}), + + receive + {'basic.ack',_,_} -> ok + after 10000 -> + error(timeout_waiting_for_initial_ack) + end, + kill_the_queue(QueueName, Config), + + amqp_channel:call(Ch, #'basic.publish'{routing_key = QueueName}, + #amqp_msg{payload = <<"HI">>}), + + receive + {'basic.ack',_,_} -> error(expecting_nack_got_ack); + {'basic.nack',_,_,_} -> ok + after 10000 -> + error(timeout_waiting_for_nack) + end. + +mixed_dead_alive_queues_reject(Config) -> + Conn = ?config(conn, Config), + {ok, Ch} = amqp_connection:open_channel(Conn), + QueueNameDead = <<"mixed_dead_alive_queues_reject_dead">>, + QueueNameAlive = <<"mixed_dead_alive_queues_reject_alive">>, + ExchangeName = <<"mixed_dead_alive_queues_reject">>, + + amqp_channel:call(Ch, #'confirm.select'{}), + amqp_channel:register_confirm_handler(Ch, self()), + + amqp_channel:call(Ch, #'queue.declare'{queue = QueueNameDead, + durable = true}), + amqp_channel:call(Ch, #'queue.declare'{queue = QueueNameAlive, + durable = true}), + + amqp_channel:call(Ch, #'exchange.declare'{exchange = ExchangeName, + durable = true}), + + amqp_channel:call(Ch, #'queue.bind'{exchange = ExchangeName, + queue = QueueNameAlive, + routing_key = <<"route">>}), + + amqp_channel:call(Ch, #'queue.bind'{exchange = ExchangeName, + queue = QueueNameDead, + routing_key = <<"route">>}), + + amqp_channel:call(Ch, #'basic.publish'{exchange = ExchangeName, + routing_key = <<"route">>}, + #amqp_msg{payload = <<"HI">>}), + + receive + {'basic.ack',_,_} -> ok; + {'basic.nack',_,_,_} -> error(expecting_ack_got_nack) + after 50000 -> + error({timeout_waiting_for_initial_ack, process_info(self(), messages)}) + end, + + kill_the_queue(QueueNameDead, Config), + + amqp_channel:call(Ch, #'basic.publish'{exchange = ExchangeName, + routing_key = <<"route">>}, + #amqp_msg{payload = <<"HI">>}), + + receive + {'basic.nack',_,_,_} -> ok; + {'basic.ack',_,_} -> error(expecting_nack_got_ack) + after 50000 -> + error({timeout_waiting_for_nack, process_info(self(), messages)}) + end. confirms_rejects_conflict(Config) -> Conn = ?config(conn, Config), @@ -88,15 +207,15 @@ confirms_rejects_conflict(Config) -> false = Conn =:= Conn1, false = Ch =:= Ch1, - QueueName = <<"confirms_rejects_conflict">>, - amqp_channel:call(Ch, #'confirm.select'{}), amqp_channel:register_confirm_handler(Ch, self()), + XOverflow = ?config(overflow, Config), + QueueName = <<"confirms_rejects_conflict", "_", XOverflow/binary>>, amqp_channel:call(Ch, #'queue.declare'{queue = QueueName, durable = true, - arguments = [{<<"x-max-length">>,long,12}, - {<<"x-overflow">>,longstr,<<"reject-publish">>}] + arguments = [{<<"x-max-length">>, long, 12}, + {<<"x-overflow">>, longstr, XOverflow}] }), %% Consume 3 messages at once. Do that often. Consume = fun Consume() -> @@ -139,12 +258,14 @@ confirms_rejects_conflict(Config) -> policy_resets_to_default(Config) -> Conn = ?config(conn, Config), + {ok, Ch} = amqp_connection:open_channel(Conn), - QueueName = <<"policy_resets_to_default">>, amqp_channel:call(Ch, #'confirm.select'{}), amqp_channel:register_confirm_handler(Ch, self()), + XOverflow = ?config(overflow, Config), + QueueName = <<"policy_resets_to_default", "_", XOverflow/binary>>, amqp_channel:call(Ch, #'queue.declare'{queue = QueueName, durable = true }), @@ -152,7 +273,7 @@ policy_resets_to_default(Config) -> rabbit_ct_broker_helpers:set_policy( Config, 0, QueueName, QueueName, <<"queues">>, - [{<<"max-length">>, MaxLength}, {<<"overflow">>, <<"reject-publish">>}]), + [{<<"max-length">>, MaxLength}, {<<"overflow">>, XOverflow}]), timer:sleep(1000), @@ -256,3 +377,26 @@ clean_acks_mailbox() -> after 1000 -> done end. + +kill_the_queue(QueueName, Config) -> + rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, kill_the_queue, [QueueName]). + +kill_the_queue(QueueName) -> + [begin + {ok, Q} = rabbit_amqqueue:lookup({resource, <<"/">>, queue, QueueName}), + Pid = amqqueue:get_pid(Q), + exit(Pid, kill) + end + || _ <- lists:seq(1, 11)], + {ok, Q} = rabbit_amqqueue:lookup({resource, <<"/">>, queue, QueueName}), + Pid = amqqueue:get_pid(Q), + case is_process_alive(Pid) of + %% Try to kill it again + true -> kill_the_queue(QueueName); + false -> ok + end. + + + + + diff --git a/test/dead_lettering_SUITE.erl b/test/dead_lettering_SUITE.erl index 6fe2a3a522..fe5e91c8ed 100644 --- a/test/dead_lettering_SUITE.erl +++ b/test/dead_lettering_SUITE.erl @@ -60,13 +60,15 @@ groups() -> {dead_letter_tests, [], [ {classic_queue, [parallel], DeadLetterTests ++ [dead_letter_ttl, + dead_letter_max_length_reject_publish_dlx, dead_letter_routing_key_cycle_ttl, dead_letter_headers_reason_expired, dead_letter_headers_reason_expired_per_message]}, {mirrored_queue, [parallel], DeadLetterTests ++ [dead_letter_ttl, - dead_letter_routing_key_cycle_ttl, - dead_letter_headers_reason_expired, - dead_letter_headers_reason_expired_per_message]}, + dead_letter_max_length_reject_publish_dlx, + dead_letter_routing_key_cycle_ttl, + dead_letter_headers_reason_expired, + dead_letter_headers_reason_expired_per_message]}, {quorum_queue, [parallel], DeadLetterTests} ]} ]. @@ -381,6 +383,31 @@ dead_letter_max_length_drop_head(Config) -> _ = consume(Ch, DLXQName, [P1, P2]), consume_empty(Ch, DLXQName). +%% Another strategy: reject-publish-dlx +dead_letter_max_length_reject_publish_dlx(Config) -> + {_Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0), + QName = ?config(queue_name, Config), + DLXQName = ?config(queue_name_dlx, Config), + + declare_dead_letter_queues(Ch, Config, QName, DLXQName, + [{<<"x-max-length">>, long, 1}, + {<<"x-overflow">>, longstr, <<"reject-publish-dlx">>}]), + + P1 = <<"msg1">>, + P2 = <<"msg2">>, + P3 = <<"msg3">>, + + %% Publish 3 messages + publish(Ch, QName, [P1, P2, P3]), + %% Consume the first one from the queue (max-length = 1) + wait_for_messages(Config, [[QName, <<"1">>, <<"1">>, <<"0">>]]), + _ = consume(Ch, QName, [P1]), + consume_empty(Ch, QName), + %% Consume the dropped ones from the dead letter queue + wait_for_messages(Config, [[DLXQName, <<"2">>, <<"2">>, <<"0">>]]), + _ = consume(Ch, DLXQName, [P2, P3]), + consume_empty(Ch, DLXQName). + %% Dead letter exchange does not have to be declared when the queue is declared, but it should %% exist by the time messages need to be dead-lettered; if it is missing then, the messages will %% be silently dropped. diff --git a/test/priority_queue_SUITE.erl b/test/priority_queue_SUITE.erl index 5bac8482fa..f7e20a9fe6 100644 --- a/test/priority_queue_SUITE.erl +++ b/test/priority_queue_SUITE.erl @@ -33,7 +33,8 @@ groups() -> {cluster_size_2, [], [ ackfold, drop, - reject, + {overflow_reject_publish, [], [reject]}, + {overflow_reject_publish_dlx, [], [reject]}, dropwhile_fetchwhile, info_head_message_timestamp, matching, @@ -87,8 +88,20 @@ init_per_group(cluster_size_3, Config) -> ]), rabbit_ct_helpers:run_steps(Config1, rabbit_ct_broker_helpers:setup_steps() ++ - rabbit_ct_client_helpers:setup_steps()). - + rabbit_ct_client_helpers:setup_steps()); +init_per_group(overflow_reject_publish, Config) -> + rabbit_ct_helpers:set_config(Config, [ + {overflow, <<"reject-publish">>} + ]); +init_per_group(overflow_reject_publish_dlx, Config) -> + rabbit_ct_helpers:set_config(Config, [ + {overflow, <<"reject-publish-dlx">>} + ]). + +end_per_group(overflow_reject_publish, _Config) -> + ok; +end_per_group(overflow_reject_publish_dlx, _Config) -> + ok; end_per_group(_Group, Config) -> rabbit_ct_helpers:run_steps(Config, rabbit_ct_client_helpers:teardown_steps() ++ @@ -334,9 +347,10 @@ drop(Config) -> reject(Config) -> {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0), - Q = <<"reject-queue">>, + XOverflow = ?config(overflow, Config), + Q = <<"reject-queue-", XOverflow/binary>>, declare(Ch, Q, [{<<"x-max-length">>, long, 4}, - {<<"x-overflow">>, longstr, <<"reject-publish">>} + {<<"x-overflow">>, longstr, XOverflow} | arguments(3)]), publish(Ch, Q, [1, 2, 3, 1, 2, 3, 1, 2, 3]), %% First 4 messages are published, all others are discarded. diff --git a/test/queue_parallel_SUITE.erl b/test/queue_parallel_SUITE.erl index 632a314d21..f8a039dc65 100644 --- a/test/queue_parallel_SUITE.erl +++ b/test/queue_parallel_SUITE.erl @@ -60,8 +60,10 @@ groups() -> [ {parallel_tests, [], [ - {classic_queue, [parallel], AllTests ++ [delete_immediately_by_pid_succeeds]}, - {mirrored_queue, [parallel], AllTests ++ [delete_immediately_by_pid_succeeds]}, + {classic_queue, [parallel], AllTests ++ [delete_immediately_by_pid_succeeds, + trigger_message_store_compaction]}, + {mirrored_queue, [parallel], AllTests ++ [delete_immediately_by_pid_succeeds, + trigger_message_store_compaction]}, {quorum_queue, [parallel], AllTests ++ [delete_immediately_by_pid_fails]}, {quorum_queue_in_memory_limit, [parallel], AllTests ++ [delete_immediately_by_pid_fails]}, {quorum_queue_in_memory_bytes, [parallel], AllTests ++ [delete_immediately_by_pid_fails]} @@ -327,6 +329,27 @@ subscribe_and_multiple_ack(Config) -> rabbit_ct_client_helpers:close_channel(Ch), ok. +trigger_message_store_compaction(Config) -> + {_, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0), + QName = ?config(queue_name, Config), + declare_queue(Ch, Config, QName), + + N = 12000, + [publish(Ch, QName, [binary:copy(<<"a">>, 5000)]) || _ <- lists:seq(1, N)], + wait_for_messages(Config, [[QName, <<"12000">>, <<"12000">>, <<"0">>]]), + + AllDTags = rabbit_ct_client_helpers:consume_without_acknowledging(Ch, QName, N), + ToAck = lists:filter(fun (I) -> I > 500 andalso I < 11200 end, AllDTags), + + [amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = Tag, + multiple = false}) || Tag <- ToAck], + + %% give compaction a moment to start in and finish + timer:sleep(5000), + amqp_channel:cast(Ch, #'queue.purge'{queue = QName}), + rabbit_ct_client_helpers:close_channel(Ch), + ok. + subscribe_and_requeue_multiple_nack(Config) -> {_, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0), QName = ?config(queue_name, Config), diff --git a/test/quorum_queue_SUITE.erl b/test/quorum_queue_SUITE.erl index c23b7ac85e..1d9789fe89 100644 --- a/test/quorum_queue_SUITE.erl +++ b/test/quorum_queue_SUITE.erl @@ -323,11 +323,12 @@ declare_invalid_args(Config) -> LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}, {<<"x-max-priority">>, long, 2000}])), - ?assertExit( - {{shutdown, {server_initiated_close, 406, _}}, _}, - declare(rabbit_ct_client_helpers:open_channel(Config, Server), - LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}, - {<<"x-overflow">>, longstr, <<"reject-publish">>}])), + [?assertExit( + {{shutdown, {server_initiated_close, 406, _}}, _}, + declare(rabbit_ct_client_helpers:open_channel(Config, Server), + LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}, + {<<"x-overflow">>, longstr, XOverflow}])) + || XOverflow <- [<<"reject-publish">>, <<"reject-publish-dlx">>]], ?assertExit( {{shutdown, {server_initiated_close, 406, _}}, _}, diff --git a/test/simple_ha_SUITE.erl b/test/simple_ha_SUITE.erl index 20012b09c8..b2caff86a9 100644 --- a/test/simple_ha_SUITE.erl +++ b/test/simple_ha_SUITE.erl @@ -30,6 +30,11 @@ all() -> ]. groups() -> + RejectTests = [ + rejects_survive_stop, + rejects_survive_sigkill, + rejects_survive_policy + ], [ {cluster_size_2, [], [ rapid_redeclare, @@ -45,9 +50,8 @@ groups() -> confirms_survive_stop, confirms_survive_sigkill, confirms_survive_policy, - rejects_survive_stop, - rejects_survive_sigkill, - rejects_survive_policy + {overflow_reject_publish, [], RejectTests}, + {overflow_reject_publish_dlx, [], RejectTests} ]} ]. @@ -69,6 +73,14 @@ init_per_group(cluster_size_2, Config) -> init_per_group(cluster_size_3, Config) -> rabbit_ct_helpers:set_config(Config, [ {rmq_nodes_count, 3} + ]); +init_per_group(overflow_reject_publish, Config) -> + rabbit_ct_helpers:set_config(Config, [ + {overflow, <<"reject-publish">>} + ]); +init_per_group(overflow_reject_publish_dlx, Config) -> + rabbit_ct_helpers:set_config(Config, [ + {overflow, <<"reject-publish-dlx">>} ]). end_per_group(_, Config) -> @@ -227,12 +239,13 @@ rejects_survive(Config, DeathFun) -> Node2Channel = rabbit_ct_client_helpers:open_channel(Config, B), %% declare the queue on the master, mirrored to the two slaves - Queue = <<"test_rejects">>, + XOverflow = ?config(overflow, Config), + Queue = <<"test_rejects", "_", XOverflow/binary>>, amqp_channel:call(Node1Channel,#'queue.declare'{queue = Queue, auto_delete = false, durable = true, arguments = [{<<"x-max-length">>, long, 1}, - {<<"x-overflow">>, longstr, <<"reject-publish">>}]}), + {<<"x-overflow">>, longstr, XOverflow}]}), Payload = <<"there can be only one">>, amqp_channel:call(Node1Channel, #'basic.publish'{routing_key = Queue}, |