Optimise msg_store recovery in case of large message file

Since 4.0.0 (commit d45fbc3) the shared message store writes large
messages into their own rdq files. This information can be utilised
when scanning rdq files during recovery to avoid reading in the whole
message body into memory unnecessarily.

This commit addresses the same issue that was addressed in 3.13.x by
commit baeefbe (ie. appending a large binary together from 4MB chunks
leaves a lot of garbage and memory fragmentation behind) but even more
efficiently.

Large messages which were written before 4.0.0, which don't fully fill
the rdq file, are still handled as before.

Author: gomoripeti

deps/rabbit/src/rabbit_msg_store.erl

@@ -1537,6 +1537,39 @@ scan_data(<<Size:64, MsgIdAndMsg:Size/binary, 255, Rest/bits>> = Data,
                     scan_data(Rest2, Fd, Fun, Offset + 1, FileSize, MsgIdsFound, Acc)
             end
     end;
+%% Large message alone in its own file
+scan_data(<<Size:64, MsgIdInt:128, _Rest/bits>> = Data, Fd, Fun, Offset, FileSize, _MsgIdsFound, _Acc)
+  when Size >= ?LARGE_MESSAGE_THRESHOLD + 16,
+       Offset == 0,
+       FileSize == Size + 9 ->
+    {ok, CurrentPos} = file:position(Fd, cur),
+    case file:pread(Fd, FileSize - 1, 1) of
+        {ok, <<255>>} ->
+            %% Avoid sub-binary construction.
+            MsgId = <<MsgIdInt:128>>,
+            TotalSize = FileSize,
+            case Fun({MsgId, TotalSize, Offset}) of
+                %% Confirmed to be a message by the provided fun.
+                {valid, Entry} ->
+                    [Entry];
+                %% Confirmed to be a message but we don't need it anymore.
+                previously_valid ->
+                    [];
+                %% Not a message, try the next byte.
+                invalid ->
+                    {ok, _} = file:position(Fd, CurrentPos),
+                    <<_, Rest2/bits>> = Data,
+                    scan_data(Rest2, Fd, Fun, Offset + 1, FileSize, #{}, [])
+            end;
+        _ ->
+            %% It is quite unexpected that FileSize matches the
+            %% expected Size but the end marker is wrong.
+            %% There is no compaction in case of large message files
+            %% so it can only happen with byte corruption?
+            <<_, Rest2/bits>> = Data,
+            {ok, _} = file:position(Fd, CurrentPos),
+            scan_data(Rest2, Fd, Fun, Offset + 1, FileSize, #{}, [])
+    end;
 %% This might be the start of a message.
 scan_data(<<Size:64, Rest/bits>> = Data, Fd, Fun, Offset, FileSize, MsgIdsFound, Acc)
           when byte_size(Rest) < Size + 1, Size < FileSize - Offset ->