fix: gracefully close message store on corrupt segments (#1710)

## Summary
- Skip async replicator close path when no followers are connected
(startup), eliminating the race condition where a spawned fiber closes
MFiles out from under the constructor
- Skip post-load processing (`load_stats_from_segments`,
`delete_unused_segments`, etc.) when the store was closed during
`load_segments_from_disk`
- Break out of `load_stats_from_segments` loop after `produce_metadata`
calls `close`, preventing access to already-closed MFiles

Fixes #1701

Author: viktorerlingsson

spec/message_store_spec.cr

@@ -231,4 +231,57 @@ describe LavinMQ::MessageStore do
       end
     end
   end
+
+  it "closes gracefully when segment has corrupt schema version with replicator" do
+    with_etcd do
+      mktmpdir do |dir|
+        # Create a valid store with a message, then close it
+        store = LavinMQ::MessageStore.new(dir, nil)
+        store.push(LavinMQ::Message.new("ex", "rk", "body"))
+        store.close
+
+        # Corrupt the schema version at the start of the segment file
+        seg_file = Dir.children(dir).find!(&.starts_with?("msgs."))
+        File.open(File.join(dir, seg_file), "r+") { |f| f.write("abcd".to_slice) }
+
+        # With a replicator (no followers), close spawns a fiber that races
+        # with the constructor — this should close gracefully, not crash
+        replicator = LavinMQ::Clustering::Server.new(LavinMQ::Config.instance, LavinMQ::Etcd.new("localhost:12379"), 0)
+        begin
+          store = LavinMQ::MessageStore.new(dir, replicator)
+          store.closed.should be_true
+        ensure
+          replicator.close
+        end
+      end
+    end
+  end
+
+  it "closes gracefully when a middle segment is corrupt with replicator" do
+    with_etcd do
+      mktmpdir do |dir|
+        # Create a store with multiple segments (one message per segment)
+        store = LavinMQ::MessageStore.new(dir, nil)
+        msg_size = LavinMQ::Config.instance.segment_size.to_u64 - (LavinMQ::BytesMessage::MIN_BYTESIZE + 5)
+        msg = LavinMQ::Message.new(RoughTime.unix_ms, "e", "k", AMQ::Protocol::Properties.new, msg_size, IO::Memory.new("a" * msg_size))
+        3.times { store.push(msg) }
+        store.@segments.size.should eq 3
+        store.close
+
+        # Corrupt the schema version of the second segment
+        seg_files = Dir.children(dir).select(&.starts_with?("msgs.")).sort!
+        File.open(File.join(dir, seg_files[1]), "r+") { |f| f.write("abcd".to_slice) }
+
+        # With a replicator (no followers), this should close gracefully
+        # even though valid segments before the corrupt one were already loaded
+        replicator = LavinMQ::Clustering::Server.new(LavinMQ::Config.instance, LavinMQ::Etcd.new("localhost:12379"), 0)
+        begin
+          store = LavinMQ::MessageStore.new(dir, replicator)
+          store.closed.should be_true
+        ensure
+          replicator.close
+        end
+      end
+    end
+  end
 end

src/lavinmq/message_store.cr

@@ -33,15 +33,18 @@ module LavinMQ
       @durable = durable
       @acks = Hash(UInt32, MFile).new { |acks, seg| acks[seg] = open_ack_file(seg) }
       load_segments_from_disk
-      delete_orphan_ack_files
-      load_deleted_from_disk
-      load_stats_from_segments
-      delete_unused_segments
+      unless @closed
+        delete_orphan_ack_files
+        load_deleted_from_disk
+        load_stats_from_segments
+        delete_unused_segments
+      end
+
       @wfile_id = @segments.last_key
       @wfile = @segments.last_value
       @rfile_id = @segments.first_key
       @rfile = @segments.first_value
-      @empty.set empty?
+      @empty.set empty? unless @closed
     end
 
     def push(msg) : SegmentPosition
@@ -260,7 +263,7 @@ module LavinMQ
       @empty.close
       # To make sure that all replication actions for the segments
       # have finished wait for a delete action of a nonexistent file
-      if replicator = @replicator
+      if (replicator = @replicator) && !replicator.all_followers.empty?
         wg = WaitGroup.new
         replicator.delete_file(File.join(@msg_dir, "nonexistent"), wg)
         spawn(name: "wait for file deletion is replicated") do
@@ -424,7 +427,9 @@ module LavinMQ
             end
           rescue ex
             @log.error { "Could not initialize segment #{seg}, closing message store: #{ex.message}" }
+            @segments[seg] = file
             close
+            break
           end
         end
         file.pos = 4
@@ -447,6 +452,7 @@ module LavinMQ
           read_metadata_file(seg, mfile)
         rescue File::NotFoundError | MetadataError
           produce_metadata(seg, mfile)
+          break if @closed
           write_metadata_file(seg, mfile) unless seg == @segments.last_key # this segment is not full yet
         end