Aggregate byte rates at vhost level for all protocols

AMQP and MQTT clients now report bytes to vhost counters
alongside their per-connection tracking. The overview endpoint
reads byte rates from vhost stats instead of summing connections,
ensuring HTTP traffic is included.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: kickster97

src/lavinmq/amqp/client.cr

@@ -246,6 +246,7 @@ module LavinMQ
         end
         @last_sent_frame = RoughTime.instant
         @send_oct_count.add(8_u64 + frame.bytesize, :relaxed)
+        @vhost.add_send_bytes(8_u64 + frame.bytesize)
         if frame.is_a?(AMQP::Frame::Connection::CloseOk)
           return false
         end
@@ -285,6 +286,7 @@ module LavinMQ
           socket.write_bytes frame, ::IO::ByteFormat::NetworkEndian
           socket.flush if websocket
           @send_oct_count.add(8_u64 + frame.bytesize, :relaxed)
+          @vhost.add_send_bytes(8_u64 + frame.bytesize)
           # Remove BCC header to not expose it to clients.
           # Table#delete will always make the underlying IO writable, even if
           # key doesn't exists. Therefore we do the has_key? check to not
@@ -300,6 +302,7 @@ module LavinMQ
           socket.write_bytes header, ::IO::ByteFormat::NetworkEndian
           socket.flush if websocket
           @send_oct_count.add(8_u64 + header.bytesize, :relaxed)
+          @vhost.add_send_bytes(8_u64 + header.bytesize)
           pos = 0
           while pos < msg.bodysize
             length = Math.min(msg.bodysize - pos, @max_frame_size - 8).to_u32
@@ -315,6 +318,7 @@ module LavinMQ
             socket.write_bytes body, ::IO::ByteFormat::NetworkEndian
             socket.flush if websocket
             @send_oct_count.add(8_u64 + body.bytesize, :relaxed)
+            @vhost.add_send_bytes(8_u64 + body.bytesize)
             pos += length
           end
           socket.flush if flush && !websocket # Websockets need to send one frame per WS frame
@@ -390,6 +394,7 @@ module LavinMQ
       private def process_frame(frame) : Nil
         @last_recv_frame = RoughTime.instant
         @recv_oct_count.add(8_u64 + frame.bytesize, :relaxed)
+        @vhost.add_recv_bytes(8_u64 + frame.bytesize)
         case frame
         when AMQP::Frame::Channel::Open
           open_channel(frame)

src/lavinmq/http/controller/main.cr

@@ -44,11 +44,6 @@ module LavinMQ
               connections += 1
               channels += c.channels.size
               consumers += c.channels.each_value.sum &.consumers.size
-              stats_details = c.stats_details
-              recv_rate += stats_details[:recv_oct_details][:rate]
-              send_rate += stats_details[:send_oct_details][:rate]
-              add_logs!(recv_rate_log, stats_details[:recv_oct_details][:log])
-              add_logs!(send_rate_log, stats_details[:send_oct_details][:log])
             end
             exchanges += vhost.exchanges.size
             queues += vhost.queues.size
@@ -59,6 +54,10 @@ module LavinMQ
               add_logs!(unacked_log, q.unacked_count_log)
             end
             vhost_stats_details = vhost.stats_details
+            recv_rate += vhost_stats_details[:recv_oct_details][:rate]
+            send_rate += vhost_stats_details[:send_oct_details][:rate]
+            add_logs!(recv_rate_log, vhost_stats_details[:recv_oct_details][:log])
+            add_logs!(send_rate_log, vhost_stats_details[:send_oct_details][:log])
             {% for sm in OVERVIEW_STATS %}
               {{ sm.id }}_count += vhost_stats_details[:{{ sm.id }}]
               {{ sm.id }}_rate += vhost_stats_details[:{{ sm.id }}_details][:rate]

src/lavinmq/mqtt/client.cr

@@ -97,6 +97,7 @@ module LavinMQ
         packet = @io.read_packet
         @log.trace { "Received packet:  #{packet.inspect}" }
         @recv_oct_count.add(packet.bytesize, :relaxed)
+        vhost.add_recv_bytes(packet.bytesize.to_u64)
 
         case packet
         when MQTT::Publish     then recieve_publish(packet)
@@ -115,6 +116,7 @@ module LavinMQ
           @io.write_packet(packet)
           @io.flush
           @send_oct_count.add(packet.bytesize, :relaxed)
+          vhost.add_send_bytes(packet.bytesize.to_u64)
         end
         case packet
         when MQTT::Publish