More reliable upgrades from old index

The files are only deleted after we have asked the message store
to sync to disk.

We always look for .idx files and load them if any, except on
clean recovery. This allows resuming the upgrade. When a node
is shut down during an upgrade (normally or not) and then
restarted, it is possible that some messages end up duplicated,
however.

Author: Loïc Hoguin

deps/rabbit/src/rabbit_classic_queue_index_v2.erl

@@ -199,29 +199,30 @@ recover(#resource{ virtual_host = VHost } = Name, Terms, IsMsgStoreClean,
              State}
     end.
 
-recover_segments(State = #mqistate { dir = Dir }, Terms, IsMsgStoreClean,
+recover_segments(State0 = #mqistate { dir = Dir }, Terms, IsMsgStoreClean,
                  ContainsCheckFun, OnSyncFun, OnSyncMsgFun, CountersRef) ->
     SegmentFiles = rabbit_file:wildcard(".*\\" ++ ?SEGMENT_EXTENSION, Dir),
-    case SegmentFiles of
-        %% No segments found. We try to see if there are segment files
-        %% from the old index.
+    State = case SegmentFiles of
+        %% No segments found.
         [] ->
-            case rabbit_file:wildcard(".*\\.idx", Dir) of
-                %% We are recovering a dirty queue that was using the old index.
-                [_|_] ->
-                    recover_index_v1_dirty(State, Terms, IsMsgStoreClean,
-                                           ContainsCheckFun, OnSyncFun, OnSyncMsgFun,
-                                           CountersRef);
-                %% Otherwise keep default values.
-                [] ->
-                    State
-            end;
+            State0;
         %% Count unackeds in the segments.
         _ ->
             Segments = lists:sort([
                 list_to_integer(filename:basename(F, ?SEGMENT_EXTENSION))
             || F <- SegmentFiles]),
-            recover_segments(State, ContainsCheckFun, CountersRef, Segments)
+            recover_segments(State0, ContainsCheckFun, CountersRef, Segments)
+    end,
+    %% We always try to see if there are segment files from the old index as well.
+    case rabbit_file:wildcard(".*\\.idx", Dir) of
+        %% We are recovering a dirty queue that was using the old index.
+        [_|_] ->
+            recover_index_v1_dirty(State, Terms, IsMsgStoreClean,
+                                   ContainsCheckFun, OnSyncFun, OnSyncMsgFun,
+                                   CountersRef);
+        %% Otherwise keep default values.
+        [] ->
+            State
     end.
 
 recover_segments(State, _, _, []) ->
@@ -384,7 +385,7 @@ recover_index_v1_loop(State0 = #mqistate{ queue_name = Name },
         MsgId = case MsgOrId of
             Msg = #basic_message{ id = MsgId0 } ->
                 %% We must do a synchronous write to avoid overloading the message store.
-                rabbit_msg_store:sync_write(MsgId0, Msg, MSClient),
+                rabbit_msg_store:blocking_write(MsgId0, Msg, MSClient),
                 MsgId0;
             MsgId0 ->
                 MsgId0
@@ -395,6 +396,7 @@ recover_index_v1_loop(State0 = #mqistate{ queue_name = Name },
         publish(MsgId, SeqId, Props, IsPersistent, IsDelivered, infinity, State1)
     end, State0, Messages),
     State = flush(State2),
+    rabbit_msg_store:force_sync(MSClient),
     %% We have written everything to disk. We can delete the old segment file
     %% to free up much needed space, to avoid doubling disk usage during the upgrade.
     rabbit_queue_index:delete_segment_file_for_seq_id(LoSeqId, V1State),
@@ -997,17 +999,17 @@ queue_index_walker_reader(#resource{ virtual_host = VHost } = Name, Gatherer) ->
     ?DEBUG("~0p ~0p", [Name, Gatherer]),
     VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
     Dir = queue_dir(VHostDir, Name),
+    SegmentFiles = rabbit_file:wildcard(".*\\" ++ ?SEGMENT_EXTENSION, Dir),
+    _ = [queue_index_walker_segment(filename:join(Dir, F), Gatherer) || F <- SegmentFiles],
     %% When there are files belonging to the old index, we go through
-    %% the old index walker function. We will upgrade to the new index
-    %% in the recover step.
+    %% the old index walker function as well.
     case rabbit_file:wildcard(".*\\.idx", Dir) of
         [_|_] ->
             rabbit_queue_index:queue_index_walker_reader(Name, Gatherer);
         [] ->
-            SegmentFiles = rabbit_file:wildcard(".*\\" ++ ?SEGMENT_EXTENSION, Dir),
-            _ = [queue_index_walker_segment(filename:join(Dir, F), Gatherer) || F <- SegmentFiles],
-            ok = gatherer:finish(Gatherer)
-    end.
+            ok
+    end,
+    ok = gatherer:finish(Gatherer).
 
 queue_index_walker_segment(F, Gatherer) ->
     ?DEBUG("~0p ~0p", [F, Gatherer]),

deps/rabbit/src/rabbit_msg_store.erl

@@ -20,7 +20,7 @@
 -export([scan_file_for_valid_messages/1]). %% salvage tool
 
 -export([transform_dir/3, force_recovery/2]). %% upgrade
--export([sync_write/3]). %% Used when upgrading to the modern index.
+-export([blocking_write/3, force_sync/1]). %% Used when upgrading to the modern index.
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
          code_change/3, prioritise_call/4, prioritise_cast/3,
@@ -516,17 +516,23 @@ write_flow(MsgId, Msg,
 
 write(MsgId, Msg, CState) -> client_write(MsgId, Msg, noflow, CState).
 
--spec sync_write(rabbit_types:msg_id(), msg(), client_msstate()) -> 'ok'.
+-spec blocking_write(rabbit_types:msg_id(), msg(), client_msstate()) -> 'ok'.
 
 %% Used when upgrading to the modern index.
-sync_write(MsgId, Msg,
-           CState = #client_msstate { cur_file_cache_ets = CurFileCacheEts,
-                                      client_ref         = CRef }) ->
+blocking_write(MsgId, Msg,
+            CState = #client_msstate { cur_file_cache_ets = CurFileCacheEts,
+                                       client_ref         = CRef }) ->
     file_handle_cache_stats:update(msg_store_write),
     ok = client_update_flying(+1, MsgId, CState),
     ok = update_msg_cache(CurFileCacheEts, MsgId, Msg),
     ok = server_call(CState, {write, CRef, MsgId, noflow}).
 
+-spec force_sync(client_msstate()) -> 'ok'.
+
+%% Used when upgrading to the modern index.
+force_sync(CState) ->
+    ok = server_call(CState, force_sync).
+
 -spec read(rabbit_types:msg_id(), client_msstate()) ->
                      {rabbit_types:ok(msg()) | 'not_found', client_msstate()}.
 
@@ -884,7 +890,11 @@ handle_call({contains, MsgId}, From, State) ->
 %% Used when upgrading to the modern index.
 handle_call(Write = {write, _, _, _}, _From, State) ->
     {noreply, State1, _} = handle_cast(Write, State),
-    reply(ok, State1).
+    reply(ok, State1);
+
+handle_call(force_sync, _From, State = #msstate { current_file_handle = CurHdl }) ->
+    ok = file_handle_cache:sync(CurHdl),
+    reply(ok, State).
 
 handle_cast({client_dying, CRef},
             State = #msstate { dying_clients       = DyingClients,