Fix rmem page released while a later chunk still points into it

_msgpack_buffer_chunk_malloc hands the current rmem page over to the
chunk that carves the reclaimed space out of it, but b->rmem_owner was
always an alias of &b->tail.mem: _msgpack_buffer_add_new_chunk copies
b->tail into a new chunk without repointing the owner, so the transfer
degenerated into a self-assignment that immediately nulled the pointer.
The page stayed owned by the older chunk instead, and because chunks are
destroyed head first, it was returned to the process global pool while
the rebuilt tail was still reading from it. Another buffer allocating a
page then received the very same memory: reads returned that buffer's
bytes and further writes corrupted it.

Let the owner follow the page when the tail is copied, so the newest
chunk carved out of a page is the one that releases it.

Also drop the carve window in _msgpack_buffer_shift_chunk when the
chunk owning the current page is destroyed. rmem_last / rmem_end kept
pointing into that page, so a later small write could carve a new chunk
out of a page that was already back in the pool.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: Watson1978

ext/msgpack/buffer.c

@@ -127,6 +127,15 @@ void msgpack_buffer_mark(void *ptr)
 
 bool _msgpack_buffer_shift_chunk(msgpack_buffer_t* b)
 {
+    if(b->rmem_owner == &b->head->mem) {
+        /* the chunk that owns the rmem page is going away and takes the page
+         * with it. don't carve the remaining space out of a page that is
+         * already back in the pool. */
+        b->rmem_end = NULL;
+        b->rmem_last = NULL;
+        b->rmem_owner = NULL;
+    }
+
     _msgpack_buffer_chunk_destroy(b->head);
 
     if(b->head == &b->tail) {
@@ -264,6 +273,18 @@ static inline msgpack_buffer_chunk_t* _msgpack_buffer_alloc_new_chunk(msgpack_bu
     return chunk;
 }
 
+/* b->tail is copied into nc and then rebuilt, so the rmem page pointer moves
+ * to nc. the owner has to follow it: if it kept pointing at b->tail.mem, the
+ * transfer in _msgpack_buffer_chunk_malloc would degenerate into a
+ * self-assignment and the page would be released by nc while the rebuilt tail
+ * is still reading from it. */
+static inline void _msgpack_buffer_transfer_rmem_owner(msgpack_buffer_t* b, msgpack_buffer_chunk_t* nc)
+{
+    if(b->rmem_owner == &b->tail.mem) {
+        b->rmem_owner = &nc->mem;
+    }
+}
+
 static inline void _msgpack_buffer_add_new_chunk(msgpack_buffer_t* b)
 {
     if(b->head == &b->tail) {
@@ -275,6 +296,7 @@ static inline void _msgpack_buffer_add_new_chunk(msgpack_buffer_t* b)
         msgpack_buffer_chunk_t* nc = _msgpack_buffer_alloc_new_chunk(b);
 
         *nc = b->tail;
+        _msgpack_buffer_transfer_rmem_owner(b, nc);
         b->head = nc;
         nc->next = &b->tail;
 
@@ -295,6 +317,7 @@ static inline void _msgpack_buffer_add_new_chunk(msgpack_buffer_t* b)
 
         /* rebuild tail */
         *nc = b->tail;
+        _msgpack_buffer_transfer_rmem_owner(b, nc);
         before_tail->next = nc;
         nc->next = &b->tail;
     }

spec/cruby/buffer_spec.rb

@@ -620,4 +620,49 @@
 
     expect(b1.read_all).to eq(('C' * 128).b)
   end
+
+  it "keeps an rmem page alive while a later chunk still points into it" do
+    threshold = 1024
+    long = 'L' * (threshold * 2)
+
+    b1 = MessagePack::Buffer.new(nil, write_reference_threshold: threshold)
+    b1.write('a' * 100)  # first rmem page
+    b1.write(long)       # written by reference
+    b1.write('b' * 10)   # second rmem page, left mostly unused
+    b1.write(long)       # reclaims the unused part of that page
+    b1.write('c' * 50)   # carved out of the reclaimed part
+    b1.read(100 + long.bytesize + 10 + long.bytesize)
+
+    # the pool must not hand that page to anyone else while b1 still reads from it
+    others = 6.times.map do |i|
+      b = MessagePack::Buffer.new
+      b.write(('A'.ord + i).chr * 4000)
+      b
+    end
+
+    expect(b1.read_all).to eq(('c' * 50).b)
+    expect(others.size).to eq(6)
+  end
+
+  it "does not carve a new chunk out of an rmem page it already released" do
+    threshold = 1024
+    long = 'L' * (threshold * 2)
+
+    b1 = MessagePack::Buffer.new(nil, write_reference_threshold: threshold)
+    b1.write('a' * 100)
+    b1.write(long)
+    b1.write('b' * 10)
+    b1.write(long)                     # reclaims the unused part of the page
+    b1.read(100 + long.bytesize + 10)  # releases that very page
+    b1.write('d' * 50)                 # must not be carved out of it
+
+    others = 6.times.map do |i|
+      b = MessagePack::Buffer.new
+      b.write(('a'.ord + i).chr * 4000)
+      b
+    end
+
+    expect(b1.read_all).to eq((long + ('d' * 50)).b)
+    expect(others.size).to eq(6)
+  end
 end