Fix RecyclerBytesStreamOutput corrupting when ending write on page boundary (#95114) (#96322)

Fixes the fact that we don't put the current page offset to zero when we
already have capacity in the buffer but are not on the last page because
of a seek.

Author: original-brownbear

docs/changelog/95114.yaml

@@ -0,0 +1,5 @@
+pr: 95114
+summary: Fix `RecyclerBytesStreamOutput` corrupting when ending write on page boundary
+area: Network
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/common/io/stream/RecyclerBytesStreamOutput.java

@@ -144,13 +144,6 @@ public void writeWithSizePrefix(Writeable writeable) throws IOException {
         }
     }
 
-    public void reset() {
-        Releasables.close(pages);
-        pages.clear();
-        pageIndex = -1;
-        currentPageOffset = pageSize;
-    }
-
     @Override
     public void flush() {
         // nothing to do
@@ -235,11 +228,11 @@ private void ensureCapacityFromPosition(long newPosition) {
             assert pageSize == newPage.v().length;
             pages.add(newPage);
             // We are at the end of the current page, increment page index
-            if (currentPageOffset == pageSize) {
-                pageIndex++;
-                currentPageOffset = 0;
-            }
             currentCapacity += pageSize;
         }
+        if (currentPageOffset == pageSize) {
+            pageIndex++;
+            currentPageOffset = 0;
+        }
     }
 }

server/src/test/java/org/elasticsearch/common/io/stream/RecyclerBytesStreamOutputTests.java

@@ -712,6 +712,22 @@ public void testReadWriteGeoPoint() throws IOException {
         }
     }
 
+    public void testWriteLongToCompletePage() throws IOException {
+        try (RecyclerBytesStreamOutput out = new RecyclerBytesStreamOutput(recycler)) {
+            out.seek(PageCacheRecycler.BYTE_PAGE_SIZE + 1);
+            int longPos = PageCacheRecycler.BYTE_PAGE_SIZE - Long.BYTES;
+            out.seek(longPos);
+            long longValue = randomLong();
+            out.writeLong(longValue);
+            byte byteValue = randomByte();
+            out.writeByte(byteValue);
+            var input = out.bytes().streamInput();
+            assertEquals(longPos, input.skip(longPos));
+            assertEquals(longValue, input.readLong());
+            assertEquals(byteValue, input.readByte());
+        }
+    }
+
     private static class TestWriteable implements Writeable {
 
         private boolean value;