Fix "Data sizes violation" error (#5492)

Under certain conditions a follower/readonlyreplica can fail to join a cluster with a "Data sizes violation" error.

This PR fixes the error condition, after which the node is able to join the cluster again.

There is no risk to the integrity of the data.

Co-authored-by: Timothy Coleman <[email protected]>

Author: github-actions[bot]

src/KurrentDB.Core.Tests/TransactionLog/when_opening_existing_ongoing_tfchunk.cs

@@ -0,0 +1,62 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+using System;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using KurrentDB.Core.TransactionLog.Chunks;
+using KurrentDB.Core.TransactionLog.Chunks.TFChunk;
+using KurrentDB.Core.TransactionLog.LogRecords;
+using KurrentDB.Core.Transforms;
+using NUnit.Framework;
+
+namespace KurrentDB.Core.Tests.TransactionLog;
+
+[TestFixture]
+public class when_opening_existing_ongoing_tfchunk : SpecificationWithFilePerTestFixture {
+	private TFChunk _chunk;
+	private TFChunk _testChunk;
+
+	[OneTimeSetUp]
+	public override async Task TestFixtureSetUp() {
+		await base.TestFixtureSetUp();
+		_chunk = await TFChunkHelper.CreateNewChunk(Filename, chunkSize: 10 * 1024);
+		_chunk.TryClose();
+		_testChunk = await TFChunk.FromOngoingFile(
+			fileSystem: new ChunkLocalFileSystem(Path.GetDirectoryName(Filename)),
+			filename: Filename,
+			writePosition: 0,
+			unbuffered: false,
+			writethrough: false,
+			reduceFileCachePressure: false, tracker: new TFChunkTracker.NoOp(),
+			getTransformFactory: DbTransformManager.Default,
+			token: CancellationToken.None);
+	}
+
+	[OneTimeTearDown]
+	public override void TestFixtureTearDown() {
+		_chunk.Dispose();
+		_testChunk.Dispose();
+		base.TestFixtureTearDown();
+	}
+
+	[Test]
+	public void the_chunk_is_cached() {
+		Assert.IsTrue(_testChunk.IsCached);
+	}
+
+	[Test]
+	public void the_chunk_is_not_readonly() {
+		Assert.IsFalse(_testChunk.IsReadOnly);
+	}
+
+	[Test]
+	// a flush before write can be triggered in practice when joining a cluster and the first
+	// replication message (8k) does not contain a full transaction.
+	public async Task can_flush_and_then_write() {
+		await _testChunk.Flush(CancellationToken.None);
+		var result = await _testChunk.TryAppend(new CommitLogRecord(0, Guid.NewGuid(), 0, DateTime.UtcNow, 0), CancellationToken.None);
+		Assert.IsTrue(result.Success);
+	}
+}

src/KurrentDB.Core.Tests/TransactionLog/when_opening_existing_tfchunk.cs

@@ -28,7 +28,7 @@ public override async Task TestFixtureSetUp() {
 			getTransformFactory: DbTransformManager.Default);
 	}
 
-	[TearDown]
+	[OneTimeTearDown]
 	public override void TestFixtureTearDown() {
 		_chunk.Dispose();
 		_testChunk.Dispose();

src/KurrentDB.Plugins/Transforms/ChunkDataWriteStream.cs

@@ -24,6 +24,7 @@ public override void Write(ReadOnlySpan<byte> buffer) {
 			ReadAndChecksum(count);
 
 			Debug.Assert(ChunkFileStream.Position == count);
+			ResetPoolingBufferedStreamReadBuffer(count);
 			_positionToHash = null;
 		}
 
@@ -48,6 +49,7 @@ private async ValueTask WriteAndChecksumAsync(long count, ReadOnlyMemory<byte> b
 		await ReadAndChecksumAsync(count, token);
 
 		Debug.Assert(ChunkFileStream.Position == count);
+		ResetPoolingBufferedStreamReadBuffer(count);
 		await ChunkFileStream.WriteAsync(buffer, token);
 		checksumAlgorithm.AppendData(buffer.Span);
 		_positionToHash = null;
@@ -95,4 +97,15 @@ private void ReadAndChecksum(long count) {
 			checksumAlgorithm.AppendData(buffer.Slice(0, bytesRead));
 		}
 	}
+
+	// Workaround PoolingBufferedStream behavior. If we leave the stream with bytes in its read buffer
+	// then when we flush it will discard those bytes and advance the Position to end of the read buffer.
+	// This would then result in attempts to write records to the wrong place in the chunk
+	// (which would be detected and rejected as a "Data sizes violation").
+	// Instead we flush here and restore the position so that the stream is set up for regular use.
+	// This can be removed if/when the PoolingBufferedStream flush behavior is updated.
+	private void ResetPoolingBufferedStreamReadBuffer(long position) {
+		ChunkFileStream.Flush();
+		ChunkFileStream.Position = position;
+	}
 }