Skip periodic flush if disk written to recently

This is meant to both reduce the number of flushes when they
are not necessary (fast writes scenarios) but also to prevent
issues on Windows when recently deleted files are not immediately
available for reopening and writing (they are in "DELETE PENDING").

Author: lhoguin

deps/rabbit/src/rabbit_classic_queue_store_v2.erl

@@ -99,7 +99,15 @@
     %% we are using file:pread/3 which leaves the file
     %% position undetermined.
     read_segment = undefined :: undefined | non_neg_integer(),
-    read_fd = undefined :: undefined | file:fd()
+    read_fd = undefined :: undefined | file:fd(),
+
+    %% Time at which the most recent buffer flush was done,
+    %% or the most recent segment delete. This is used to skip
+    %% the periodic flush when a flush was done recently, or
+    %% when the current write file was recently deleted (to
+    %% avoid {error,eacces} issues on Windows for files in
+    %% "DELETE PENDING" state).
+    last_disk_write = erlang:monotonic_time(millisecond)
 }).
 
 -type state() :: #qs{}.
@@ -184,9 +192,15 @@ get_write_offset(Segment, Size, State = #qs{ write_segment = undefined }) ->
 
 -spec sync(State) -> State when State::state().
 
-sync(State) ->
+sync(State=#qs{ last_disk_write = LastDiskWrite }) ->
     ?DEBUG("~0p", [State]),
-    flush_buffer(State, fun(_) -> ok end).
+    case erlang:monotonic_time(millisecond) of
+        %% Half the sync interval in rabbitmq_amqqueue_process.
+        Now when Now > LastDiskWrite + 100 ->
+            flush_buffer(State, fun(_) -> ok end);
+        _ ->
+            State
+    end.
 
 maybe_flush_buffer(State = #qs{ write_buffer_size = WriteBufferSize }) ->
     case WriteBufferSize >= max_cache_size() of
@@ -227,7 +241,8 @@ flush_buffer(State0 = #qs{ write_buffer = WriteBuffer }, FsyncFun) ->
     %% Finally we move the write_buffer to the cache.
     State#qs{ write_buffer = #{},
               write_buffer_size = 0,
-              cache = WriteBuffer }.
+              cache = WriteBuffer,
+              last_disk_write = erlang:monotonic_time(millisecond) }.
 
 flush_buffer_build(WriteBuffer = [{FirstSeqId, {Offset, _, _}}|_],
                    CheckCRC32, SegmentEntryCount) ->
@@ -548,7 +563,8 @@ delete_segments(Segments, State0 = #qs{ write_buffer = WriteBuffer0,
             end
     end, {#{}, WriteBufferSize0}, WriteBuffer0),
     State#qs{ write_buffer = WriteBuffer,
-              write_buffer_size = WriteBufferSize }.
+              write_buffer_size = WriteBufferSize,
+              last_disk_write = erlang:monotonic_time(millisecond) }.
 
 %% ----
 %%