perf(serialization): optimize buffered tail reads

src/Orleans.Serialization/Buffers/ArcBufferWriter.cs

@@ -378,12 +378,24 @@ internal bool TryPeek(long offset, out byte value)
             throw new ArgumentOutOfRangeException(nameof(offset));
         }
 
-        if (offset >= Length)
+        var length = Length;
+        if (offset >= length)
         {
             value = default;
             return false;
         }
 
+        // Owned buffers end at the write page's written length; pinned slices can end partway through a page.
+        if (!_hasPinnedPages)
+        {
+            var writePageOffset = offset - (length - _writePage.Length);
+            if (writePageOffset >= 0)
+            {
+                value = _writePage.AsSpan((int)writePageOffset, 1)[0];
+                return true;
+            }
+        }
+
         Debug.Assert(offset <= int.MaxValue);
         var remainingOffset = (int)offset;
         var current = _readPage;

test/Orleans.Journaling.Tests/JournalBufferWriterOwnershipTests.cs

@@ -1,6 +1,7 @@
 using System.Buffers;
 using System.Text;
 using Orleans.Journaling.Json;
+using Orleans.Serialization.Buffers;
 using Xunit;
 
 namespace Orleans.Journaling.Tests;
@@ -254,6 +255,50 @@ public void GetEntryByte_RejectsReadsOutsideActiveEntry()
         Assert.Empty(ToArray(writer));
     }
 
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public void JsonEmptyEntry_RetainsValidationAfterLargeOrConsumedPrefix(bool consumePrefix)
+    {
+        using var writer = CreateWriter(JsonFormat);
+        var payload = Encoding.UTF8.GetBytes($"[\"set\",\"{new string('x', ArcBufferWriter.MinimumPageSize * 3 + 17)}\"]");
+        AppendEntry(writer.CreateJournalStreamWriter(new JournalStreamId(1)), payload);
+        using var prefix = writer.GetCommittedBuffer();
+        var prefixBytes = Encoding.UTF8.GetBytes($"[1,{Encoding.UTF8.GetString(payload)}]\n");
+        Assert.Equal(prefixBytes, prefix.ToArray());
+        if (consumePrefix)
+        {
+            writer.Consume(prefix);
+        }
+
+        var exception = Assert.Throws<InvalidOperationException>(() =>
+            AppendEntry(writer.CreateJournalStreamWriter(new JournalStreamId(2)), []));
+        Assert.Equal("The JSON Lines journal entry has no entry payload.", exception.Message);
+        Assert.Equal(consumePrefix ? [] : prefixBytes, ToArray(writer));
+
+        AppendEntry(writer.CreateJournalStreamWriter(new JournalStreamId(3)), GetThirdPayload(JsonFormat));
+        var suffix = """[3,["set",3]]""" + "\n";
+        Assert.Equal(Encoding.UTF8.GetBytes((consumePrefix ? "" : Encoding.UTF8.GetString(prefixBytes)) + suffix), ToArray(writer));
+        Assert.Equal(prefixBytes, prefix.ToArray());
+    }
+
+    [Fact]
+    public void JsonEmptyEntry_RetainsValidationOnNewAndResetWriter()
+    {
+        using var writer = CreateWriter(JsonFormat);
+        for (var iteration = 0; iteration < 2; iteration++)
+        {
+            var stream = writer.CreateJournalStreamWriter(new JournalStreamId(1));
+            var exception = Assert.Throws<InvalidOperationException>(() => AppendEntry(stream, []));
+            Assert.Equal("The JSON Lines journal entry has no entry payload.", exception.Message);
+            Assert.Empty(ToArray(writer));
+
+            AppendEntry(stream, GetFirstPayload(JsonFormat));
+            Assert.Equal(Encoding.UTF8.GetBytes("""[1,["set",1]]""" + "\n"), ToArray(writer));
+            writer.Reset();
+        }
+    }
+
     private static JournalBufferWriter CreateWriter(string format) => format switch
     {
         BinaryFormat => new OrleansBinaryJournalBufferWriter(),

test/Orleans.Serialization.UnitTests/ArcBufferWriterTests.cs

@@ -142,6 +142,117 @@ public void AsReadOnlySequence_SingleSegment_DoesNotAllocate()
         Assert.Equal(data.Length * 1_000L, length);
     }
 
+    [Theory]
+    [InlineData(0)]
+    [InlineData(17)]
+    [InlineData(PageSize)]
+    [InlineData(PageSize * 2 + 3)]
+    [InlineData(PageSize * 3 + 3)]
+    public void Peek_ReturnsEveryUnreadByteAcrossPages(int consumed)
+    {
+        using var writer = new ArcBufferWriter();
+        var data = Enumerable.Range(0, PageSize * 3 + 17).Select(static i => (byte)(i % 251)).ToArray();
+        writer.Write(data);
+        writer.AdvanceReader(consumed);
+
+        for (var offset = 0; offset < writer.Length; offset++)
+        {
+            Assert.Equal(data[consumed + offset], writer.Reader.Peek(offset));
+        }
+
+        Assert.Throws<ArgumentOutOfRangeException>(() => writer.Reader.Peek(-1));
+        Assert.Throws<ArgumentOutOfRangeException>(() => writer.Reader.Peek(writer.Length));
+        Assert.Throws<ArgumentOutOfRangeException>(() => writer.Reader.Peek(long.MaxValue));
+    }
+
+    [Fact]
+    public void Peek_UsesWrittenPageWhenLaterPagesAreReserved()
+    {
+        using var writer = new ArcBufferWriter();
+        var buffers = new List<ArraySegment<byte>>(4);
+        writer.ReplenishBuffers(buffers);
+        var data = Enumerable.Range(0, PageSize + 17).Select(static i => (byte)(i % 251)).ToArray();
+        writer.Write(data);
+
+        Assert.Equal(data[^1], writer.Reader.Peek(writer.Length - 1));
+        Assert.Equal(data[PageSize], writer.Reader.Peek(PageSize));
+        Assert.Equal(data[0], writer.Reader.Peek(0));
+    }
+
+    [Fact]
+    public void Peek_PreservesTailWhenNewWritePageIsEmpty()
+    {
+        using var writer = new ArcBufferWriter();
+        writer.Write([1, 2, 3]);
+        var memory = writer.GetMemory(PageSize * 2);
+
+        Assert.Equal(3, writer.Reader.Peek(2));
+        Assert.Throws<ArgumentOutOfRangeException>(() => writer.Reader.Peek(3));
+
+        memory.Span[0] = 4;
+        writer.AdvanceWriter(1);
+        Assert.Equal(4, writer.Reader.Peek(3));
+        Assert.Equal(3, writer.Reader.Peek(2));
+    }
+
+    [Fact]
+    public void Peek_TracksTruncationPatchingAndAppends()
+    {
+        using var writer = new ArcBufferWriter();
+        var data = Enumerable.Range(0, PageSize * 3 + 17).Select(static i => (byte)(i % 251)).ToArray();
+        writer.Write(data);
+        writer.AdvanceReader(13);
+        writer.Truncate(PageSize + 7);
+
+        Assert.Equal(data[13 + PageSize + 6], writer.Reader.Peek(writer.Length - 1));
+        writer.WriteAt(writer.Length - 1, [255]);
+        Assert.Equal(255, writer.Reader.Peek(writer.Length - 1));
+        writer.Write([254, 253]);
+        Assert.Equal(254, writer.Reader.Peek(writer.Length - 2));
+        Assert.Equal(253, writer.Reader.Peek(writer.Length - 1));
+    }
+
+    [Fact]
+    public void Peek_RespectsPinnedSliceEndWhenSourceTailGrows()
+    {
+        using var source = new ArcBufferWriter();
+        var data = Enumerable.Range(0, PageSize + 64).Select(static i => (byte)(i % 251)).ToArray();
+        source.Write(data);
+        using var all = source.PeekSlice(source.Length);
+        using var slice = all.Slice(13, PageSize + 4);
+        using var destination = new ArcBufferWriter();
+        destination.AppendPinned(slice);
+        source.Write([253, 254, 255]);
+
+        for (var offset = 0; offset < destination.Length; offset++)
+        {
+            Assert.Equal(data[13 + offset], destination.Reader.Peek(offset));
+        }
+
+        destination.AdvanceReader(PageSize);
+        Assert.Equal(data[PageSize + 16], destination.Reader.Peek(3));
+        Assert.Throws<ArgumentOutOfRangeException>(() => destination.Reader.Peek(4));
+    }
+
+    [Fact]
+    public void Peek_ObservesResetReuseAndDisposal()
+    {
+        using var writer = new ArcBufferWriter();
+        var reader = writer.Reader;
+        writer.Write([1, 2, 3]);
+        writer.AdvanceReader(3);
+        Assert.Throws<ArgumentOutOfRangeException>(() => reader.Peek(0));
+        writer.Write([6]);
+        Assert.Equal(6, reader.Peek(0));
+
+        writer.Reset();
+        Assert.Throws<ArgumentOutOfRangeException>(() => reader.Peek(0));
+        writer.Write([4, 5]);
+        Assert.Equal(5, reader.Peek(1));
+        writer.Dispose();
+        Assert.Throws<ObjectDisposedException>(() => reader.Peek(0));
+    }
+
     [Fact]
     public void AppendPinned_WholeBufferPinsPagesAndReadsData()
     {