Remove dangling symlinks on log init.

Author: kjnilsson

src/ra_kv_harness.erl

@@ -531,6 +531,9 @@ execute_operation(State, {kill_member}) ->
 
             Pid = erpc:call(NodeName, erlang, whereis, [?CLUSTER_NAME]),
             erpc:call(NodeName, erlang, exit, [Pid, kill]),
+            %% give it a bit of time after a kill in case this member is chosen
+            %% for the next operation
+            timer:sleep(100),
             State#{operations_count => OpCount + 1,
                    successful_ops => SuccessOps + 1}
     end.

src/ra_log.erl

@@ -229,6 +229,7 @@ init(#{uid := UId,
     % this queries the segment writer and thus blocks until any
     % segments it is currently processed have been finished
     MtRange = ra_mt:range(Mt0),
+    ok = ra_log_segments:purge_dangling_symlinks(Dir),
     SegRefs = my_segrefs(UId, SegWriter),
     SegmentMaxCount = maps:get(segment_max_entries, Conf, ?SEGMENT_MAX_ENTRIES),
     SegmentMaxSize = maps:get(segment_max_size_bytes, Conf, ?SEGMENT_MAX_SIZE_B),
@@ -838,7 +839,6 @@ handle_event({segments, TidRanges, NewSegs},
 
     %% it is theoretically possible that the segment writer flush _could_
     %% over take WAL notifications
-    %%
     FstPend = ra_seq:first(Pend0),
     Pend = case ra_mt:range(Mt) of
                {Start, _End} when Start > FstPend ->
@@ -1372,7 +1372,7 @@ release_resources(MaxOpenSegments, AccessPattern,
 %% only used by resend to wal functionality and doesn't update the mem table
 wal_rewrite(#?MODULE{cfg = #cfg{uid = UId,
                                 wal = Wal} = Cfg,
-                     last_wal_write = {_, _, _LastWalIdx}} = State,
+                     last_wal_write = {_, _, _}} = State,
             Tid, {Idx, Term, Cmd}) ->
     case ra_log_wal:write(Wal, {UId, self()}, Tid, Idx, Term, Cmd) of
         {ok, Pid} ->

src/ra_log_segment.erl

@@ -495,13 +495,13 @@ info(Filename, Live0)
   when not is_tuple(Filename) ->
     %% TODO: this can be much optimised by a specialised index parsing
     %% function
-    {ok, Seg} = open(Filename, #{mode => read}),
-    Index = Seg#state.index,
-    {ok, #file_info{type = T,
+    {ok, #file_info{type = Type,
                     links = Links,
-                    ctime = CTime}} = file:read_link_info(Filename,
-                                                          [raw, {time, posix}]),
+                    ctime = CTime}} = prim_file:read_link_info(Filename,
+                                                               [raw, {time, posix}]),
 
+    {ok, Seg} = open(Filename, #{mode => read}),
+    Index = Seg#state.index,
     AllIndexesSeq = ra_seq:from_list(maps:keys(Index)),
     Live = case Live0 of
                undefined ->
@@ -515,7 +515,7 @@ info(Filename, Live0)
                            end, 0, Live),
     Info = #{size => Seg#state.data_write_offset,
              index_size => Seg#state.data_start,
-             file_type => T,
+             file_type => Type,
              links => Links,
              ctime => CTime,
              max_count => max_count(Seg),

src/ra_log_segments.erl

@@ -29,6 +29,7 @@
          fetch_term/2,
          info/1,
          purge_symlinks/2,
+         purge_dangling_symlinks/1,
          compaction_conf/1
          ]).
 
@@ -394,7 +395,7 @@ purge_symlinks(Dir, OlderThanSec) ->
     Now = erlang:system_time(second),
     [begin
          Fn = filename:join(Dir, F),
-         case file:read_link_info(Fn, [raw, {time, posix}]) of
+         case prim_file:read_link_info(Fn, [raw, {time, posix}]) of
              {ok, #file_info{type = symlink,
                              ctime = Time}}
                when Now - Time > OlderThanSec ->
@@ -406,6 +407,24 @@ purge_symlinks(Dir, OlderThanSec) ->
      end || F <- list_files(Dir, ".segment")],
     ok.
 
+-spec purge_dangling_symlinks(file:filename_all()) -> ok.
+purge_dangling_symlinks(Dir) ->
+    [begin
+         Fn = filename:join(Dir, F),
+         case file:read_link_info(Fn, [raw]) of
+             {ok, #file_info{type = symlink}} ->
+                 case file:open(Fn, [raw, read, binary]) of
+                     {ok, Fd} ->
+                         ok = file:close(Fd);
+                     {error, enoent} ->
+                         %% dangling symlink
+                         ok = prim_file:delete(Fn)
+                 end;
+             _ ->
+                 ok
+         end
+     end || F <- list_files(Dir, ".segment")],
+    ok.
 %% LOCAL
 
 segment_read_plan(_SegRefs, [], Acc) ->