Introduce delete_object and use it to solve race

3 files changed, 23 insertions, 19 deletions

diff --git a/include/rabbit_msg_store_index.hrl b/include/rabbit_msg_store_index.hrl
index fba0b7cd..d4115363 100644
--- a/include/rabbit_msg_store_index.hrl
+++ b/include/rabbit_msg_store_index.hrl
@@ -51,6 +51,7 @@
                                                [{fieldpos(), fieldvalue()}]),
                           index_state()) -> 'ok').
 -spec(delete/2 :: (rabbit_guid:guid(), index_state()) -> 'ok').
+-spec(delete_object/2 :: (keyvalue(), index_state()) -> 'ok').
 -spec(delete_by_file/2 :: (fieldvalue(), index_state()) -> 'ok').
 -spec(terminate/1 :: (index_state()) -> any()).
 
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index 7b715b80..8acd05e7 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -1583,7 +1583,7 @@ combine_files(#file_summary { file             = Source,
     %%   copy back in, and then copy over from Source
     %% otherwise we just truncate straight away and copy over from Source
     {DestinationWorkList, DestinationValid} =
-        find_unremoved_messages_in_file(Destination, State),
+        load_and_vacuum_message_file(Destination, State),
     {DestinationContiguousTop, DestinationWorkListTail} =
         drop_contiguous_block_prefix(DestinationWorkList),
     case DestinationWorkListTail of
@@ -1609,8 +1609,7 @@ combine_files(#file_summary { file             = Source,
               ok = file_handle_cache:sync(DestinationHdl),
               ok = file_handle_cache:delete(TmpHdl)
     end,
-    {SourceWorkList, SourceValid} =
-        find_unremoved_messages_in_file(Source, State),
+    {SourceWorkList, SourceValid} = load_and_vacuum_message_file(Source, State),
     ok = copy_messages(SourceWorkList, DestinationValid, ExpectedSize,
                        SourceHdl, DestinationHdl, Destination, State),
     %% tidy up
@@ -1618,25 +1617,25 @@ combine_files(#file_summary { file             = Source,
     ok = file_handle_cache:delete(SourceHdl),
     ExpectedSize.
 
-find_unremoved_messages_in_file(File,
-                                {_FileSummaryEts, Dir, Index, IndexState}) ->
+load_and_vacuum_message_file(File, {_FileSummaryEts, Dir, Index, IndexState}) ->
     %% Messages here will be end-of-file at start-of-list
     {ok, Messages, _FileSize} =
         scan_file_for_valid_messages(Dir, filenum_to_name(File)),
     %% foldl will reverse so will end up with msgs in ascending offset order
-    lists:foldl(fun ({Guid, TotalSize, Offset}, Acc = {List, Size}) ->
-                        case Index:lookup(Guid, IndexState) of
-                            #msg_location { file = File, total_size = TotalSize,
-                                            offset = Offset, ref_count = 0 } ->
-                                ok = Index:delete(Guid, IndexState),
-                                Acc;
-                            #msg_location { file = File, total_size = TotalSize,
-                                            offset = Offset } = Entry ->
-                                {[ Entry | List ], TotalSize + Size};
-                            _ ->
-                                Acc
-                        end
-                end, {[], 0}, Messages).
+    lists:foldl(
+      fun ({Guid, TotalSize, Offset}, Acc = {List, Size}) ->
+              case Index:lookup(Guid, IndexState) of
+                  #msg_location { file = File, total_size = TotalSize,
+                                  offset = Offset, ref_count = 0 } = Entry ->
+                      ok = Index:delete_object(Entry, IndexState),
+                      Acc;
+                  #msg_location { file = File, total_size = TotalSize,
+                                  offset = Offset } = Entry ->
+                      {[ Entry | List ], TotalSize + Size};
+                  _ ->
+                      Acc
+              end
+      end, {[], 0}, Messages).
 
 copy_messages(WorkList, InitOffset, FinalOffset, SourceHdl, DestinationHdl,
               Destination, {_FileSummaryEts, _Dir, Index, IndexState}) ->
diff --git a/src/rabbit_msg_store_ets_index.erl b/src/rabbit_msg_store_ets_index.erl
index 1eb3c11f..79d8b62b 100644
--- a/src/rabbit_msg_store_ets_index.erl
+++ b/src/rabbit_msg_store_ets_index.erl
@@ -35,7 +35,7 @@
 
 -export([new/1, recover/1,
          lookup/2, insert/2, update/2, update_fields/3, delete/2,
-         delete_by_file/2, terminate/1]).
+         delete_object/2, delete_by_file/2, terminate/1]).
 
 -define(MSG_LOC_NAME, rabbit_msg_store_ets_index).
 -define(FILENAME, "msg_store_index.ets").
@@ -79,6 +79,10 @@ delete(Key, State) ->
     true = ets:delete(State #state.table, Key),
     ok.
 
+delete_object(Obj, State) ->
+    true = ets:match_delete(State #state.table, Obj),
+    ok.
+
 delete_by_file(File, State) ->
     MatchHead = #msg_location { file = File, _ = '_' },
     ets:select_delete(State #state.table, [{MatchHead, [], [true]}]),