Fix: reset pos when crossing segment boundary in find_offset_in_segments (#1772)

### WHAT is this pull request doing?

Fixes #1771

Restores the `pos = 4u32` reset in `find_offset_in_segments` when
advancing to the next segment.

In #1479, the original `pos = 4_u32` was changed to `rfile.pos = 4u32`.
During
[review](#1479 (comment))
it was mentioned that the method uses a local `pos` variable, not
`rfile.pos`. When resolving, the line was removed entirely instead of
being changed back to `pos = 4u32`.

Without the reset, when a timestamp-based `x-stream-offset` falls
between two segments, the method walks all messages in the first
segment, crosses to the next segment, and reads from the old segment's
end position, causing an `IndexError` or potentially a SEGFAULT.

  ### HOW can this pull request be tested?

  Run added spec

---------

Co-authored-by: Viktor Erlingsson <[email protected]>

Author: kickster97

spec/stream_queue_spec.cr

@@ -221,6 +221,32 @@ describe LavinMQ::AMQP::Stream do
     end
   end
 
+  it "consume from timestamp offset across segment boundary" do
+    with_amqp_server do |s|
+      with_channel(s) do |ch|
+        q = ch.queue("stream-ts-across-segments", args: stream_queue_args)
+        # Use half-segment messages so exactly 1 fits per segment
+        data = Bytes.new(LavinMQ::Config.instance.segment_size // 2)
+        # Fill segment 1 — two half-segment messages won't fit, so second triggers new segment
+        q.publish_confirm data
+        # Sleep to create a timestamp gap
+        sleep 1.seconds
+        target_time = Time.utc
+        # This publish can't fit in current segment, creates a new one with first_ts > target
+        q.publish_confirm data
+        # Consume from timestamp in the gap — find_offset_in_segments must cross segment boundary
+        ch.prefetch 1
+        msgs = Channel(AMQP::Client::DeliverMessage).new
+        q.subscribe(no_ack: false, args: AMQP::Client::Arguments.new({"x-stream-offset": target_time})) do |msg|
+          msgs.send msg
+          msg.ack
+        end
+        msg = msgs.receive
+        StreamSpecHelpers.offset_from_headers(msg.properties.headers).should eq 2
+      end
+    end
+  end
+
   describe "Expiration" do
     it "segments should be removed if max-length set" do
       with_amqp_server do |s|

src/lavinmq/amqp/stream/stream_message_store.cr

@@ -105,6 +105,7 @@ module LavinMQ::AMQP
         if rfile.nil? || pos == rfile.size
           if segment = @segments.each_key.find { |sid| sid > segment }
             rfile = @segments[segment]
+            pos = 4u32
             msg_offset = @segment_first_offset[segment]
           else
             return last_offset_seg_pos