Expose some message index related functions for salvage tool

`rabbit_msg_store:scan_file_for_valid_messages/1`
`rabbit_queue_index:scan_queue_segments/4`

(cherry picked from commit 924df7f)

Conflicts:
	deps/rabbit/src/rabbit_queue_index.erl

Author: HoloRin

deps/rabbit/src/rabbit_msg_store.erl

@@ -17,6 +17,8 @@
 -export([set_maximum_since_use/2, combine_files/3,
          delete_file/2]). %% internal
 
+-export([scan_file_for_valid_messages/1]). %% salvage tool
+
 -export([transform_dir/3, force_recovery/2]). %% upgrade
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
@@ -1392,12 +1394,15 @@ should_mask_action(CRef, MsgId,
 %% file helper functions
 %%----------------------------------------------------------------------------
 
-open_file(Dir, FileName, Mode) ->
+open_file(File, Mode) ->
     file_handle_cache:open_with_absolute_path(
-      form_filename(Dir, FileName), ?BINARY_MODE ++ Mode,
+      File, ?BINARY_MODE ++ Mode,
       [{write_buffer, ?HANDLE_CACHE_BUFFER_SIZE},
        {read_buffer,  ?HANDLE_CACHE_BUFFER_SIZE}]).
 
+open_file(Dir, FileName, Mode) ->
+    open_file(form_filename(Dir, FileName), Mode).
+
 close_handle(Key, CState = #client_msstate { file_handle_cache = FHC }) ->
     CState #client_msstate { file_handle_cache = close_handle(Key, FHC) };
 
@@ -1687,18 +1692,22 @@ recover_crashed_compaction(Dir, TmpFileName, NonTmpRelatedFileName) ->
     ok = file_handle_cache:delete(TmpHdl),
     ok.
 
-scan_file_for_valid_messages(Dir, FileName) ->
-    case open_file(Dir, FileName, ?READ_MODE) of
+scan_file_for_valid_messages(File) ->
+    case open_file(File, ?READ_MODE) of
         {ok, Hdl}       -> Valid = rabbit_msg_file:scan(
-                                     Hdl, filelib:file_size(
-                                            form_filename(Dir, FileName)),
+                                     Hdl, filelib:file_size(File),
                                      fun scan_fun/2, []),
                            ok = file_handle_cache:close(Hdl),
                            Valid;
         {error, enoent} -> {ok, [], 0};
-        {error, Reason} -> {error, {unable_to_scan_file, FileName, Reason}}
+        {error, Reason} -> {error, {unable_to_scan_file,
+                                    filename:basename(File),
+                                    Reason}}
     end.
 
+scan_file_for_valid_messages(Dir, FileName) ->
+    scan_file_for_valid_messages(form_filename(Dir, FileName)).
+
 scan_fun({MsgId, TotalSize, Offset, _Msg}, Acc) ->
     [{MsgId, TotalSize, Offset} | Acc].
 

deps/rabbit/src/rabbit_queue_index.erl

@@ -16,7 +16,7 @@
          read/3, next_segment_boundary/1, bounds/1, start/2, stop/1]).
 
 -export([add_queue_ttl/0, avoid_zeroes/0, store_msg_size/0, store_msg/0]).
--export([scan_queue_segments/3]).
+-export([scan_queue_segments/3, scan_queue_segments/4]).
 
 %% Migrates from global to per-vhost message stores
 -export([move_to_per_vhost_stores/1,
@@ -255,8 +255,9 @@
 
 -spec erase(rabbit_amqqueue:name()) -> 'ok'.
 
-erase(Name) ->
-    #qistate { dir = Dir } = blank_state(Name),
+erase(#resource{ virtual_host = VHost } = Name) ->
+    VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
+    #qistate { dir = Dir } = blank_state(VHostDir, Name),
     erase_index_dir(Dir).
 
 %% used during variable queue purge when there are no pending acks
@@ -278,8 +279,9 @@ reset_state(#qistate{ queue_name     = Name,
 -spec init(rabbit_amqqueue:name(),
                  on_sync_fun(), on_sync_fun()) -> qistate().
 
-init(Name, OnSyncFun, OnSyncMsgFun) ->
-    State = #qistate { dir = Dir } = blank_state(Name),
+init(#resource{ virtual_host = VHost } = Name, OnSyncFun, OnSyncMsgFun) ->
+    VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
+    State = #qistate { dir = Dir } = blank_state(VHostDir, Name),
     false = rabbit_file:is_file(Dir), %% is_file == is file or dir
     State#qistate{on_sync     = OnSyncFun,
                   on_sync_msg = OnSyncMsgFun}.
@@ -290,9 +292,10 @@ init(Name, OnSyncFun, OnSyncMsgFun) ->
                         {'undefined' | non_neg_integer(),
                          'undefined' | non_neg_integer(), qistate()}.
 
-recover(Name, Terms, MsgStoreRecovered, ContainsCheckFun,
-        OnSyncFun, OnSyncMsgFun) ->
-    State = blank_state(Name),
+recover(#resource{ virtual_host = VHost } = Name, Terms, MsgStoreRecovered,
+        ContainsCheckFun, OnSyncFun, OnSyncMsgFun) ->
+    VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
+    State = blank_state(VHostDir, Name),
     State1 = State #qistate{on_sync     = OnSyncFun,
                             on_sync_msg = OnSyncMsgFun},
     CleanShutdown = Terms /= non_clean_shutdown,
@@ -549,17 +552,16 @@ erase_index_dir(Dir) ->
         false -> ok
     end.
 
-blank_state(QueueName) ->
-    Dir = queue_dir(QueueName),
+blank_state(VHostDir, QueueName) ->
+    Dir = queue_dir(VHostDir, QueueName),
     blank_state_name_dir_funs(QueueName,
                          Dir,
                          fun (_) -> ok end,
                          fun (_) -> ok end).
 
-queue_dir(#resource{ virtual_host = VHost } = QueueName) ->
+queue_dir(VHostDir, QueueName) ->
     %% Queue directory is
     %% {node_database_dir}/msg_stores/vhosts/{vhost}/queues/{queue}
-    VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
     QueueDir = queue_name_to_dir_name(QueueName),
     filename:join([VHostDir, "queues", QueueDir]).
 
@@ -725,9 +727,13 @@ queue_index_walker_reader(QueueName, Gatherer) ->
            end, ok, QueueName),
     ok = gatherer:finish(Gatherer).
 
-scan_queue_segments(Fun, Acc, QueueName) ->
+scan_queue_segments(Fun, Acc, #resource{ virtual_host = VHost } = QueueName) ->
+    VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
+    scan_queue_segments(Fun, Acc, VHostDir, QueueName).
+
+scan_queue_segments(Fun, Acc, VHostDir, QueueName) ->
     State = #qistate { segments = Segments, dir = Dir } =
-        recover_journal(blank_state(QueueName)),
+        recover_journal(blank_state(VHostDir, QueueName)),
     Result = lists:foldr(
       fun (Seg, AccN) ->
               segment_entries_foldr(
@@ -1465,12 +1471,13 @@ drive_transform_fun(Fun, Hdl, Contents) ->
                                drive_transform_fun(Fun, Hdl, Contents1)
     end.
 
-move_to_per_vhost_stores(#resource{} = QueueName) ->
+move_to_per_vhost_stores(#resource{virtual_host = VHost} = QueueName) ->
     OldQueueDir = filename:join([queues_base_dir(), "queues",
                                  queue_name_to_dir_name_legacy(QueueName)]),
-    NewQueueDir = queue_dir(QueueName),
+    VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
+    NewQueueDir = queue_dir(VHostDir, QueueName),
     _ = rabbit_log_upgrade:info("About to migrate queue directory '~s' to '~s'",
-                            [OldQueueDir, NewQueueDir]),
+                                [OldQueueDir, NewQueueDir]),
     case rabbit_file:is_dir(OldQueueDir) of
         true  ->
             ok = rabbit_file:ensure_dir(NewQueueDir),

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -382,6 +382,7 @@ start_queue(Config) ->
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
     LQ = ?config(queue_name, Config),
+    Children = length(rpc:call(Server, supervisor, which_children, [?SUPNAME])),
     ?assertEqual({'queue.declare_ok', LQ, 0, 0},
                  declare(Ch, LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
 
@@ -404,6 +405,7 @@ start_queue(Config) ->
     %% Check that the application and process are still up
     ?assertMatch({ra, _, _}, lists:keyfind(ra, 1,
                                            rpc:call(Server, application, which_applications, []))),
+    Expected = Children + 1,
     ?assertMatch(Expected,
                  length(rpc:call(Server, supervisor, which_children, [?SUPNAME]))),