Fix delayed message store crash on corrupt segment data (#1694)

## Summary
- Fix crash on startup when delayed message store segment files are
corrupt or truncated (e.g. after unclean shutdown)
- Rewrite `build_index` to iterate segments directly with
`IO::EOFError`/`IndexError` handling, matching how `produce_metadata`
handles the same scenario in the base `MessageStore`
- Add two specs covering truncated segments (with acked messages) and
trailing corrupt data (with valid messages preserved)

Closes #1693

## Test plan
- [x] Spec: rebuild index with truncated segment and acked messages
- [x] Spec: rebuild index preserving valid messages despite trailing
corrupt data
- [ ] Run full test suite (`make test`)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: viktorerlingsson

spec/delayed_message_exchange_spec.cr

@@ -60,6 +60,55 @@ describe "Delayed Message Exchange" do
         wait_for { s.vhosts["/"].queues[delay_q_name].message_count == 0 }
       end
     end
+
+    it "should rebuild index on restart with truncated segment and acked messages" do
+      with_amqp_server do |s|
+        with_channel(s) do |ch|
+          x = ch.exchange(x_name, "topic", args: x_args)
+          q = ch.queue("delayed_q")
+          q.bind(x.name, "#")
+          hdrs = AMQP::Client::Arguments.new({"x-delay" => 1})
+          x.publish_confirm "test message", "rk", props: AMQP::Client::Properties.new(headers: hdrs)
+          wait_for { s.vhosts["/"].queues["delayed_q"].message_count == 1 }
+        end
+        # All messages in delayed queue are now expired and acked
+        s.vhosts["/"].queues[delay_q_name].message_count.should eq 0
+        data_dir = File.join(s.vhosts["/"].data_dir, Digest::SHA1.hexdigest(delay_q_name))
+
+        s.stop
+
+        # Truncate segment file mid-message, simulating a crash where
+        # mmap data wasn't fully flushed to disk
+        seg_file = File.join(data_dir, "msgs.0000000001")
+        File.open(seg_file, "r+") { |f| f.truncate(f.size // 2) }
+
+        s.restart
+        s.vhosts["/"].queues[delay_q_name].message_count.should eq 0
+      end
+    end
+
+    it "should rebuild index on restart preserving valid messages despite trailing corrupt data" do
+      with_amqp_server do |s|
+        with_channel(s) do |ch|
+          x = ch.exchange(x_name, "topic", args: x_args)
+          # No consumer bound, so message stays in the delayed queue
+          hdrs = AMQP::Client::Arguments.new({"x-delay" => 300_000})
+          x.publish_confirm "test message", "rk", props: AMQP::Client::Properties.new(headers: hdrs)
+          s.vhosts["/"].queues[delay_q_name].message_count.should eq 1
+        end
+        data_dir = File.join(s.vhosts["/"].data_dir, Digest::SHA1.hexdigest(delay_q_name))
+
+        s.stop
+
+        # Append a partial message (just a non-zero timestamp, no body),
+        # simulating a crash mid-write
+        seg_file = File.join(data_dir, "msgs.0000000001")
+        File.open(seg_file, "a") { |f| f.write_bytes(1i64, IO::ByteFormat::SystemEndian) }
+
+        s.restart
+        s.vhosts["/"].queues[delay_q_name].message_count.should eq 1
+      end
+    end
   end
 
   q_name = "delayed_q"

src/lavinmq/amqp/queue/delayed_exchange_queue/delayed_message_store.cr

@@ -36,10 +36,27 @@ module LavinMQ::AMQP
         requeued = DelayedRequeuedStore.new
         count = 0u32
         bytesize = 0u64
-        while env = shift?
-          count += 1
-          bytesize += env.message.bytesize
-          requeued.insert(env.segment_position, env.message.timestamp)
+        @segments.each do |seg_id, mfile|
+          mfile.pos = 4
+          loop do
+            pos = mfile.pos.to_u32
+            break if pos == mfile.size
+            if deleted?(seg_id, pos)
+              BytesMessage.skip(mfile)
+              next
+            end
+            ts = IO::ByteFormat::SystemEndian.decode(Int64, mfile.to_slice(pos, 8))
+            break if ts.zero? # This means that the rest of the file is zero, so break.
+            msg = BytesMessage.from_bytes(mfile.to_slice + pos)
+            sp = SegmentPosition.make(seg_id, pos, msg)
+            mfile.seek(sp.bytesize, IO::Seek::Current)
+            count += 1
+            bytesize += msg.bytesize
+            requeued.insert(sp, msg.timestamp)
+          rescue ex : IO::EOFError | IndexError
+            @log.warn(exception: ex) { "Corrupt data at segment #{seg_id} pos #{pos}, skipping rest of segment" }
+            break
+          end
         end
         @requeued = requeued
         @size = count