Don't get stuck on index key with no messages

src/rabbit_delayed_message.erl

@@ -29,7 +29,9 @@
 -export([messages_delayed/1]).
 
 %% For testing, debugging and manual use
--export([refresh_config/0]).
+-export([refresh_config/0,
+         table_name/0,
+         index_table_name/0]).
 
 -import(rabbit_delayed_message_utils, [swap_delay_header/1]).
 
@@ -141,7 +143,7 @@ handle_cast(_C, State) ->
 handle_info({timeout, _TimerRef, {deliver, Key}}, State) ->
     case mnesia:dirty_read(?TABLE_NAME, Key) of
         [] ->
-            ok;
+            mnesia:dirty_delete(?INDEX_TABLE_NAME, Key);
         Deliveries ->
             route(Key, Deliveries, State),
             mnesia:dirty_delete(?TABLE_NAME, Key),
@@ -162,7 +164,7 @@ maybe_delay_first() ->
     case mnesia:dirty_first(?INDEX_TABLE_NAME) of
         %% destructuring to prevent matching '$end_of_table'
         #delay_key{timestamp = FirstTS} = Key2 ->
-            %% there are messages that expired and need to be delivered
+            %% there are messages that will expire and need to be delivered
             Now = erlang:system_time(milli_seconds),
             start_timer(FirstTS - Now, Key2);
         _ ->
@@ -314,3 +316,9 @@ bump_routed_stats(ExName, Qs, State) ->
 
 refresh_config(State) ->
     rabbit_event:init_stats_timer(State, #state.stats_state).
+
+table_name() ->
+    ?TABLE_NAME.
+
+index_table_name() ->
+    ?INDEX_TABLE_NAME.

test/plugin_SUITE.erl

@@ -32,6 +32,7 @@ groups() ->
                                 delayed_messages_count,
                                 node_restart_before_delay_expires,
                                 node_restart_after_delay_expires,
+                                no_message_for_index,
                                 string_delay_header
                                ]},
      {fine_stats, [], [
@@ -327,6 +328,37 @@ node_restart_after_delay_expires(Config) ->
 
     ok.
 
+no_message_for_index(Config) ->
+    Chan = rabbit_ct_client_helpers:open_channel(Config),
+
+    Ex = make_exchange_name(Config, "1"),
+    Q = make_queue_name(Config, "1"),
+
+    setup_fabric(Chan, make_durable_exchange(Ex, <<"direct">>),
+                 make_durable_queue(Q)),
+
+    Msgs = [1000, 2000],
+
+    publish_messages(Chan, Ex, Msgs),
+
+    %% delete the first message, but keep it in the index table
+    make_table_corrupted(Config),
+
+    timer:sleep(2000),
+
+    %% the index key with no messages should be ignored/deleted
+    %% and the following messages are successfully delivered
+    Msgs2 = tl(Msgs),
+    {ok, Result} = consume(Chan, Q, Msgs2),
+    ?assertEqual(Msgs2, Result),
+
+    amqp_channel:call(Chan, #'exchange.delete' { exchange = Ex }),
+    amqp_channel:call(Chan, #'queue.delete' { queue = Q }),
+
+    rabbit_ct_client_helpers:close_channel(Chan),
+
+    ok.
+
 string_delay_header(Config) ->
     Chan = rabbit_ct_client_helpers:open_channel(Config),
 
@@ -478,3 +510,10 @@ set_collect_stats(Config, CollectStats) ->
 
 refresh_config(Config) ->
     ok = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_delayed_message, refresh_config, []).
+
+make_table_corrupted(Config) ->
+    Table = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_delayed_message, table_name, []),
+    IndexTable = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_delayed_message, index_table_name, []),
+
+    FirstKey = rabbit_ct_broker_helpers:rpc(Config, 0, mnesia, dirty_first, [IndexTable]),
+    rabbit_ct_broker_helpers:rpc(Config, 0, mnesia, dirty_delete, [Table, FirstKey]).