Skip to content

Commit db1ff6c

Browse files
kjnilssonkjnilsson
committed
QQ: delay consumer metrics deletion until unsettled messages are settled
Cancelling a consumer with unacked messages used to delete its consumer_created row immediately, even though the consumer itself lingers in state until those messages are settled/discarded/returned. Move the delete effect into update_or_remove_con/5, the single place a cancelled consumer is actually dropped from state, so the metrics row disappears exactly when the consumer does rather than up front. down/remove still delete immediately, since those forcibly return all messages and remove the consumer right away.
1 parent 479f654 commit db1ff6c

2 files changed

Lines changed: 102 additions & 56 deletions

File tree

‎deps/rabbit/src/rabbit_fifo.erl‎

Lines changed: 75 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -381,8 +381,9 @@ apply_(#{index := Idx} = Meta,
381381
State1 = State0#?STATE{messages = rabbit_fifo_pq:in(?DEFAULT_PRIORITY,
382382
?MSG(Idx, Header),
383383
Messages)},
384-
State2 = update_or_remove_con(Meta, ConsumerKey, Con, State1),
385-
{State3, Effects} = activate_next_consumer({State2, []}),
384+
{State2, Effects0} = update_or_remove_con(Meta, ConsumerKey, Con,
385+
State1, []),
386+
{State3, Effects} = activate_next_consumer({State2, Effects0}),
386387
checkout(Meta, State0, State3, Effects);
387388
_ ->
388389
{State00, ok, []}
@@ -423,11 +424,11 @@ apply_(#{index := Index,
423424
{State0, {dequeue, empty}};
424425
_ ->
425426
Timeout = get_consumer_timeout(ConsumerMeta, State00),
426-
{_, State1} = update_consumer(Meta, ConsumerId, ConsumerId,
427-
ConsumerMeta,
428-
{once, {simple_prefetch, 1}}, 0,
429-
Timeout, State0),
430-
case checkout_one(Meta, false, State1, []) of
427+
{_, State1, Effects00} = update_consumer(Meta, ConsumerId, ConsumerId,
428+
ConsumerMeta,
429+
{once, {simple_prefetch, 1}},
430+
0, Timeout, State0, []),
431+
case checkout_one(Meta, false, State1, Effects00) of
431432
{success, _, MsgId, Msg, _ExpiredMsg, State2, Effects0} ->
432433
RaftIdx = get_msg_idx(Msg),
433434
Header = get_msg_header(Msg),
@@ -503,11 +504,11 @@ apply_(#{index := Idx} = Meta,
503504
error ->
504505
ConsumerId
505506
end,
506-
{Consumer, State1} = update_consumer(Meta, ConsumerKey, ConsumerId,
507-
ConsumerMeta, Spec, Priority,
508-
Timeout, State0),
507+
{Consumer, State1, Effects00} = update_consumer(Meta, ConsumerKey, ConsumerId,
508+
ConsumerMeta, Spec, Priority,
509+
Timeout, State0, []),
509510
WasActive = is_active(ConsumerKey, State1),
510-
{State2, Effs0} = activate_next_consumer(State1, []),
511+
{State2, Effs0} = activate_next_consumer(State1, Effects00),
511512
{Active, ActivityStatus} = consumer_activity_status(ConsumerKey, State2),
512513
Effs = case {WasActive, Active} of
513514
{false, true} ->
@@ -713,7 +714,7 @@ apply_(Meta, {nodeup, Node}, #?STATE{consumers = Cons0,
713714
{consumer_disconnected_timeout, ConsumerKey},
714715
infinity} | EAcc1],
715716

716-
{update_or_remove_con(Meta, ConsumerKey, C, SAcc), EAcc};
717+
update_or_remove_con(Meta, ConsumerKey, C, SAcc, EAcc);
717718
(_, _, Acc) ->
718719
Acc
719720
end, {State0, Effects0}, maps:iterator(Cons0, ordered)),
@@ -1904,16 +1905,11 @@ cancel_consumer0(Meta, ConsumerKey,
19041905
#?STATE{consumers = C0} = S0, Effects0, Reason) ->
19051906
case C0 of
19061907
#{ConsumerKey := Consumer} ->
1907-
{S, Effects2} = maybe_return_all(Meta, ConsumerKey, Consumer,
1908-
S0, Effects0, Reason),
1909-
1910-
1911-
%% The effects are emitted before the consumer is actually removed
1912-
%% if the consumer has unacked messages. This is a bit weird but
1913-
%% in line with what classic queues do (from an external point of
1914-
%% view)
1915-
Effects = cancel_consumer_effects(consumer_id(Consumer), S, Effects2),
1916-
{S, Effects};
1908+
%% the consumer_created/deleted metrics effect is emitted by
1909+
%% maybe_return_all/update_or_remove_con, only once the consumer
1910+
%% is actually removed (immediately, or once any unsettled
1911+
%% messages have all been settled/discarded/returned)
1912+
maybe_return_all(Meta, ConsumerKey, Consumer, S0, Effects0, Reason);
19171913
_ ->
19181914
%% already removed: do nothing
19191915
{S0, Effects0}
@@ -2032,18 +2028,23 @@ maybe_return_all(#{system_time := Ts} = Meta, ConsumerKey,
20322028
Effects0, Reason) ->
20332029
case Reason of
20342030
cancel ->
2035-
{update_or_remove_con(
2036-
Meta, ConsumerKey,
2037-
Consumer#consumer{cfg = CCfg#consumer_cfg{lifetime = once},
2038-
credit = 0,
2039-
status = cancelled},
2040-
S0), Effects0};
2031+
%% keep the consumer around, and its metrics untouched, until
2032+
%% any unsettled messages are settled/discarded/returned; see
2033+
%% update_or_remove_con/5
2034+
update_or_remove_con(
2035+
Meta, ConsumerKey,
2036+
Consumer#consumer{cfg = CCfg#consumer_cfg{lifetime = once},
2037+
credit = 0,
2038+
status = cancelled},
2039+
S0, Effects0);
20412040
_ ->
2041+
%% down/remove: the consumer is gone immediately regardless of
2042+
%% any unsettled messages, which are returned to the queue below
20422043
{S1, Effects} = return_all(Meta, S0, Effects0, ConsumerKey,
20432044
Consumer, Reason == down),
20442045
{S1#?STATE{consumers = maps:remove(ConsumerKey, S1#?STATE.consumers),
20452046
last_active = Ts},
2046-
Effects}
2047+
cancel_consumer_effects(consumer_id(Consumer), S1, Effects)}
20472048
end.
20482049

20492050
node_of(undefined) -> undefined;
@@ -2288,11 +2289,12 @@ complete(Meta, ConsumerKey, MsgIds,
22882289
Len = map_size(Checked0) - map_size(Checked),
22892290
Con = Con0#consumer{checked_out = Checked,
22902291
credit = increase_credit(Con0, Len)},
2291-
State1 = update_or_remove_con(Meta, ConsumerKey, Con, State0),
2292+
{State1, Effects1} = update_or_remove_con(Meta, ConsumerKey, Con, State0,
2293+
Effects),
22922294
{State1#?STATE{msg_bytes_checkout = BytesCheckout - SettledSize,
22932295
reclaimable_bytes = ReclBytes + SettledSize + (Len * ?ENQ_OVERHEAD_B),
22942296
messages_total = Tot - Len},
2295-
Effects}.
2297+
Effects1}.
22962298

22972299
increase_credit(#consumer{cfg = #consumer_cfg{lifetime = once},
22982300
credit = Credit}, _) ->
@@ -2492,8 +2494,9 @@ return_one(#{system_time := Ts} = Meta, MsgId,
24922494
false ->
24932495
State0#?STATE{returns = lqueue:in(Msg, Returns)}
24942496
end,
2495-
State = update_or_remove_con(Meta, ConsumerKey, Con, State1),
2496-
{add_bytes_return(Header, State), Effects0}
2497+
{State, Effects1} = update_or_remove_con(Meta, ConsumerKey, Con,
2498+
State1, Effects0),
2499+
{add_bytes_return(Header, State), Effects1}
24972500
end.
24982501

24992502
should_delay(DeliveryFailed, DelayedRetry, Ts, Header, Anns) ->
@@ -2851,10 +2854,11 @@ assign_to_consumer(#{system_time := Ts} = Meta, _Ts, ConsumerKey, Msgs,
28512854
State1 = State0#?STATE{msg_bytes_checkout = BytesCheckout,
28522855
msg_bytes_enqueue = BytesEnqueue,
28532856
next_consumer_timeout = NextConTimeout},
2854-
State = update_or_remove_con(Meta, ConsumerKey, Con, State1),
2857+
{State, Effects1} = update_or_remove_con(Meta, ConsumerKey, Con, State1,
2858+
Effects0),
28552859
DelMsgs = lists:reverse(DeliveryMsgs),
28562860
DeliveryEffect = delivery_effect(ConsumerKey, DelMsgs, State),
2857-
Effects = [DeliveryEffect | Effects0],
2861+
Effects = [DeliveryEffect | Effects1],
28582862
{State, Effects}.
28592863

28602864
delayed_in(ReadyAt, Idx, Msg, DeferralToken, #delayed{tree = Tree0,
@@ -2989,10 +2993,11 @@ checkout_one(#{system_time := Ts} = Meta, ExpiredMsg0, InitState0, Effects0) ->
29892993
BytesEnqueue - Size,
29902994
next_consumer_timeout =
29912995
min(Timeout, NextConTimeout)},
2992-
State = update_or_remove_con(Meta, ConsumerKey,
2993-
Con, State1),
2996+
{State, Effects2} = update_or_remove_con(
2997+
Meta, ConsumerKey, Con,
2998+
State1, Effects1),
29942999
{success, ConsumerKey, Next, Msg, ExpiredMsg,
2995-
State, Effects1}
3000+
State, Effects2}
29963001
end;
29973002
empty ->
29983003
{nochange, ExpiredMsg, InitState, Effects1}
@@ -3184,33 +3189,41 @@ update_or_remove_con(Meta, ConsumerKey,
31843189
#consumer{cfg = #consumer_cfg{lifetime = once},
31853190
checked_out = Checked,
31863191
credit = 0} = Con,
3187-
#?STATE{consumers = Cons} = State) ->
3192+
#?STATE{consumers = Cons} = State, Effects) ->
31883193
case map_size(Checked) of
31893194
0 ->
31903195
#{system_time := Ts} = Meta,
3191-
% we're done with this consumer
3192-
State#?STATE{consumers = maps:remove(ConsumerKey, Cons),
3193-
last_active = Ts};
3196+
% we're done with this consumer: this is the only point a
3197+
% lifetime=once consumer is actually removed, whether that's
3198+
% immediately on cancellation or, if it still had unsettled
3199+
% messages, once the last of them is settled/discarded/returned;
3200+
% the consumer metrics are cleared to match, not before
3201+
{State#?STATE{consumers = maps:remove(ConsumerKey, Cons),
3202+
last_active = Ts},
3203+
cancel_consumer_effects(consumer_id(Con), State, Effects)};
31943204
_ ->
3195-
% there are unsettled items so need to keep around
3196-
State#?STATE{consumers = maps:put(ConsumerKey, Con, Cons)}
3205+
% there are unsettled items so need to keep around; the
3206+
% consumer metrics are left as-is until it is actually removed
3207+
{State#?STATE{consumers = maps:put(ConsumerKey, Con, Cons)}, Effects}
31973208
end;
31983209
update_or_remove_con(_Meta, ConsumerKey,
31993210
#consumer{status = quiescing,
32003211
checked_out = Checked} = Con0,
32013212
#?STATE{consumers = Cons,
3202-
waiting_consumers = Waiting} = State)
3213+
waiting_consumers = Waiting} = State, Effects)
32033214
when map_size(Checked) == 0 ->
32043215
Con = Con0#consumer{status = up},
3205-
State#?STATE{consumers = maps:remove(ConsumerKey, Cons),
3206-
waiting_consumers = add_waiting({ConsumerKey, Con}, Waiting)};
3216+
{State#?STATE{consumers = maps:remove(ConsumerKey, Cons),
3217+
waiting_consumers = add_waiting({ConsumerKey, Con}, Waiting)},
3218+
Effects};
32073219
update_or_remove_con(_Meta, ConsumerKey,
32083220
#consumer{} = Con,
32093221
#?STATE{consumers = Cons,
3210-
service_queue = ServiceQueue} = State) ->
3211-
State#?STATE{consumers = maps:put(ConsumerKey, Con, Cons),
3222+
service_queue = ServiceQueue} = State, Effects) ->
3223+
{State#?STATE{consumers = maps:put(ConsumerKey, Con, Cons),
32123224
service_queue = maybe_queue_consumer(ConsumerKey, Con,
3213-
ServiceQueue)}.
3225+
ServiceQueue)},
3226+
Effects}.
32143227

32153228
maybe_queue_consumer(Key, #consumer{credit = Credit,
32163229
status = up,
@@ -3231,7 +3244,7 @@ maybe_queue_consumer(_Key, _Consumer, ServiceQueue) ->
32313244
update_consumer(Meta, ConsumerKey, {Tag, Pid}, ConsumerMeta,
32323245
{Life, Mode} = Spec, Priority, Timeout,
32333246
#?STATE{cfg = #cfg{consumer_strategy = competing},
3234-
consumers = Cons0} = State0) ->
3247+
consumers = Cons0} = State0, Effects0) ->
32353248
Consumer = case Cons0 of
32363249
#{ConsumerKey := #consumer{} = Consumer0} ->
32373250
merge_consumer(Meta, Consumer0, ConsumerMeta,
@@ -3249,29 +3262,35 @@ update_consumer(Meta, ConsumerKey, {Tag, Pid}, ConsumerMeta,
32493262
credit = Credit,
32503263
delivery_count = DeliveryCount}
32513264
end,
3252-
{Consumer, update_or_remove_con(Meta, ConsumerKey, Consumer, State0)};
3265+
{State, Effects} = update_or_remove_con(Meta, ConsumerKey, Consumer,
3266+
State0, Effects0),
3267+
{Consumer, State, Effects};
32533268
update_consumer(Meta, ConsumerKey, {Tag, Pid}, ConsumerMeta,
32543269
{Life, Mode} = Spec, Priority, Timeout,
32553270
#?STATE{cfg = #cfg{consumer_strategy = single_active},
32563271
consumers = Cons0,
32573272
waiting_consumers = Waiting0,
3258-
service_queue = _ServiceQueue0} = State) ->
3273+
service_queue = _ServiceQueue0} = State, Effects0) ->
32593274
%% if it is the current active consumer, just update
32603275
%% if it is a cancelled active consumer, add to waiting unless it is the only
32613276
%% one, then merge
32623277
case active_consumer(Cons0) of
32633278
{ConsumerKey, #consumer{status = up} = Consumer0} ->
32643279
Consumer = merge_consumer(Meta, Consumer0, ConsumerMeta,
32653280
Spec, Priority),
3266-
{Consumer, update_or_remove_con(Meta, ConsumerKey, Consumer, State)};
3281+
{State1, Effects} = update_or_remove_con(Meta, ConsumerKey,
3282+
Consumer, State, Effects0),
3283+
{Consumer, State1, Effects};
32673284
undefined when is_map_key(ConsumerKey, Cons0) ->
32683285
%% there is no active consumer and the current consumer is in the
32693286
%% consumers map and thus must be cancelled, in this case we can just
32703287
%% merge and effectively make this the current active one
32713288
Consumer0 = maps:get(ConsumerKey, Cons0),
32723289
Consumer = merge_consumer(Meta, Consumer0, ConsumerMeta,
32733290
Spec, Priority),
3274-
{Consumer, update_or_remove_con(Meta, ConsumerKey, Consumer, State)};
3291+
{State1, Effects} = update_or_remove_con(Meta, ConsumerKey,
3292+
Consumer, State, Effects0),
3293+
{Consumer, State1, Effects};
32753294
_ ->
32763295
%% add as a new waiting consumer
32773296
Credit = included_credit(Mode),
@@ -3286,7 +3305,7 @@ update_consumer(Meta, ConsumerKey, {Tag, Pid}, ConsumerMeta,
32863305
credit = Credit,
32873306
delivery_count = DeliveryCount},
32883307
Waiting = add_waiting({ConsumerKey, Consumer}, Waiting0),
3289-
{Consumer, State#?STATE{waiting_consumers = Waiting}}
3308+
{Consumer, State#?STATE{waiting_consumers = Waiting}, Effects0}
32903309
end.
32913310

32923311
add_waiting({Key, _} = New, Waiting) ->

‎deps/rabbit/test/rabbit_fifo_SUITE.erl‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1316,6 +1316,33 @@ cancelled_checkout_out_test(Config) ->
13161316
apply(meta(Config, 7), make_checkout(Cid, {dequeue, settled}, #{}), State4),
13171317
ok.
13181318

1319+
cancel_consumer_metrics_deletion_delayed_until_settled_test(Config) ->
1320+
Cid = {<<"cid">>, self()},
1321+
{State0, _} = enq(Config, 1, 1, first, test_init(test)),
1322+
{State1, #{key := CKey, next_msg_id := NextMsgId}, _} =
1323+
checkout(Config, ?LINE, Cid, 1, State0), %% prefetch of 1, 1 msg out
1324+
%% cancelling while a message is still checked out must not delete the
1325+
%% consumer metrics yet
1326+
{State2, _, Effects1} = apply(meta(Config, 3),
1327+
rabbit_fifo:make_checkout(Cid, cancel, #{}),
1328+
State1),
1329+
?assertNot(has_cancel_consumer_handler_effect(Effects1)),
1330+
?assertEqual(1, map_size(State2#rabbit_fifo.consumers)),
1331+
%% settling the last unacked message finally removes the consumer, and
1332+
%% only then should its metrics be deleted
1333+
{State3, ok, Effects2} =
1334+
apply(meta(Config, 4), rabbit_fifo:make_settle(CKey, [NextMsgId]),
1335+
State2),
1336+
?assertEqual(0, map_size(State3#rabbit_fifo.consumers)),
1337+
?assert(has_cancel_consumer_handler_effect(Effects2)),
1338+
ok.
1339+
1340+
has_cancel_consumer_handler_effect(Effects) ->
1341+
lists:any(fun ({mod_call, rabbit_quorum_queue,
1342+
cancel_consumer_handler, _}) -> true;
1343+
(_) -> false
1344+
end, Effects).
1345+
13191346
down_with_noproc_consumer_returns_unsettled_test(Config) ->
13201347
Cid = {?FUNCTION_NAME_B, self()},
13211348
{State0, _} = enq(Config, 1, 1, second, test_init(test)),

0 commit comments

Comments
 (0)