Fix segfault race condition when closing auto-delete queues (#1529)

## Summary
- Use WaitGroup to track active deliver loop fibers and wait for them to
complete before closing the message store
- Prevents race condition where MFile could be unmapped while a consumer
was still delivering a message (causing segfault in `properties.cr:224`
during `to_io()`)
- Reorder Consumer#close to close notification channels before calling
rm_consumer to prevent deadlock during auto_delete queue cleanup

## Background
When a heartbeat timeout triggers client cleanup on an auto-delete
queue, the following race can occur:
1. Consumer deliver loop is writing message headers to socket (which
yields)
2. Client cleanup triggers queue auto-delete → queue close → message
store close → MFile unmap
3. Consumer deliver loop resumes and tries to continue writing from
unmapped memory → SEGFAULT

The fix uses Crystal's WaitGroup to ensure all deliver loops have exited
before the message store is closed.

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

---------

Co-authored-by: Claude <[email protected]>

Author: carlhoerberg

src/lavinmq/amqp/consumer.cr

@@ -31,17 +31,18 @@ module LavinMQ
         @flow = @channel.flow?
         @metadata = @channel.@metadata.extend({consumer: @tag})
         @log = Logger.new(Log, @metadata)
-        spawn deliver_loop, name: "Consumer deliver loop"
+        fiber_name = "Consumer vhost=#{@queue.vhost.name} queue=#{@queue.name}"
+        @queue.deliver_loop_wg.spawn(name: fiber_name) { deliver_loop }
         @flow_change = BoolChannel.new(@flow)
         @has_capacity = BoolChannel.new(true)
       end
 
       def close
         @closed = true
-        @queue.rm_consumer(self)
         @notify_closed.close
         @has_capacity.close
         @flow_change.close
+        @queue.rm_consumer(self)
       end
 
       @notify_closed = ::Channel(Nil).new

src/lavinmq/amqp/queue/queue.cr

@@ -90,6 +90,7 @@ module LavinMQ::AMQP
 
     @msg_store_lock = Mutex.new(:reentrant)
     @msg_store : MessageStore
+    getter deliver_loop_wg = WaitGroup.new
 
     getter paused = BoolChannel.new(false)
 
@@ -414,7 +415,8 @@ module LavinMQ::AMQP
         @consumers.each &.cancel
         @consumers.clear
       end
-      Fiber.yield # Allow all consumers to cancel before closing mmap:s
+      Fiber.yield           # Let deliver_loop fibers start and react to closed channels
+      @deliver_loop_wg.wait # Wait for all deliver loops to exit before closing mmap:s
       @msg_store_lock.synchronize do
         @msg_store.close
       end