fix(grpc): preserve expected-version failures with unknown revisions

src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/AppendAcrossRestartGrpcTests.cs

@@ -0,0 +1,78 @@
+using System;
+using System.IO;
+using System.Linq;
+using System.Threading.Tasks;
+using EventStore.Client.Streams;
+using EventStore.Core.Tests.Helpers;
+using NUnit.Framework;
+
+namespace EventStore.Core.Tests.Services.Transport.Grpc.StreamsTests;
+
+[Category("LongRunning")]
+public class AppendAcrossRestartGrpcTests : SpecificationWithDirectoryPerTestFixture
+{
+	private MiniNode<LogFormat.V2, string> _node;
+	private GrpcStreamEdgeOperations _grpc;
+	private string _dbPath;
+
+	[OneTimeSetUp]
+	public override async Task TestFixtureSetUp()
+	{
+		await base.TestFixtureSetUp();
+		_dbPath = Path.Combine(PathName, "restart-node-db");
+		await StartNode(waitForAdminUserCreation: true);
+	}
+
+	[OneTimeTearDown]
+	public override async Task TestFixtureTearDown()
+	{
+		_grpc?.Dispose();
+		if (_node is not null)
+			await _node.Shutdown();
+		await base.TestFixtureTearDown();
+	}
+
+	[Test]
+	public async Task detects_existing_streams_and_metadata_after_restart()
+	{
+		const string stream = "grpc-existing-stream-across-restart";
+		const string metadataStream = "$$grpc-metadata-across-restart";
+		AssertSuccess(await _grpc.Append(stream, count: 10, noStream: true), 9);
+		AssertSuccess(await _grpc.Append(metadataStream, noStream: true,
+			data: "{\"$maxCount\":5}", eventType: "$metadata"), 0);
+		AssertSuccess(await _grpc.Append("grpc-last-stream-before-restart", noStream: true), 0);
+
+		await Task.Delay(500);
+		await _node.Shutdown(keepDb: true);
+		_grpc.Dispose();
+		await StartNode(waitForAdminUserCreation: false);
+
+		AssertSuccess(await _grpc.Append(stream, expectedRevision: 9), 10);
+		AssertSuccess(await _grpc.Append(metadataStream, expectedRevision: 0,
+			data: "{\"$maxCount\":6}", eventType: "$metadata"), 1);
+
+		var events = await _grpc.Read(stream, 0, 20);
+		Assert.That(events.Count(x => x.Event is not null), Is.EqualTo(11));
+		Assert.That(events.Last(x => x.Event is not null).Event.Event.StreamRevision,
+			Is.EqualTo(10));
+	}
+
+	private async Task StartNode(bool waitForAdminUserCreation)
+	{
+		_node = new MiniNode<LogFormat.V2, string>(PathName,
+			dbPath: _dbPath,
+			streamExistenceFilterSize: 10_000,
+			streamExistenceFilterCheckpointIntervalMs: 100,
+			streamExistenceFilterCheckpointDelayMs: 0);
+		await _node.Start();
+		if (waitForAdminUserCreation)
+			await _node.AdminUserCreated;
+		_grpc = new GrpcStreamEdgeOperations(_node);
+	}
+
+	private static void AssertSuccess(BatchAppendResp response, ulong expectedRevision)
+	{
+		Assert.That(response.ResultCase, Is.EqualTo(BatchAppendResp.ResultOneofCase.Success));
+		Assert.That(response.Success.CurrentRevision, Is.EqualTo(expectedRevision));
+	}
+}

src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/GrpcStreamEdgeOperations.cs

@@ -0,0 +1,152 @@
+using System;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using EventStore.Client.Streams;
+using EventStore.Core.Services.Transport.Grpc;
+using EventStore.Core.Tests.Helpers;
+using Google.Protobuf;
+using Grpc.Core;
+using Grpc.Net.Client;
+using NUnit.Framework;
+using GrpcMetadata = EventStore.Core.Services.Transport.Grpc.Constants.Metadata;
+using Streams = EventStore.Client.Streams.Streams;
+
+namespace EventStore.Core.Tests.Services.Transport.Grpc.StreamsTests;
+
+internal sealed class GrpcStreamEdgeOperations : IDisposable
+{
+	private readonly GrpcChannel _channel;
+	private readonly Streams.StreamsClient _client;
+	private readonly CallCredentials _credentials;
+	private CallOptions CallOptions => new(credentials: _credentials,
+		deadline: DateTime.UtcNow.AddSeconds(20));
+
+	public GrpcStreamEdgeOperations(MiniNode<LogFormat.V2, string> node)
+	{
+		_channel = GrpcChannel.ForAddress(new UriBuilder { Scheme = Uri.UriSchemeHttps }.Uri,
+			new GrpcChannelOptions { HttpClient = node.HttpClient, DisposeHttpClient = false });
+		_client = new Streams.StreamsClient(_channel);
+		_credentials = CallCredentials.FromInterceptor((_, metadata) =>
+		{
+			metadata.Add("authorization", "Basic " + Convert.ToBase64String(
+				Encoding.ASCII.GetBytes("admin:changeit")));
+			return Task.CompletedTask;
+		});
+	}
+
+	public GrpcStreamEdgeOperations(GrpcChannel channel)
+	{
+		_client = new Streams.StreamsClient(channel);
+		_credentials = CallCredentials.FromInterceptor((_, metadata) =>
+		{
+			metadata.Add("authorization", "Basic " + Convert.ToBase64String(
+				Encoding.ASCII.GetBytes("admin:changeit")));
+			return Task.CompletedTask;
+		});
+	}
+
+	public async Task<BatchAppendResp> Append(
+		string streamName,
+		int count = 1,
+		ulong? expectedRevision = null,
+		bool noStream = false,
+		string data = "event",
+		string eventType = "event")
+	{
+		using var call = _client.BatchAppend(CallOptions);
+		var options = new BatchAppendReq.Types.Options
+		{
+			StreamIdentifier = new() { StreamName = ByteString.CopyFromUtf8(streamName) }
+		};
+		if (expectedRevision.HasValue)
+			options.StreamPosition = expectedRevision.Value;
+		else if (noStream)
+			options.NoStream = new();
+		else
+			options.Any = new();
+
+		var request = new BatchAppendReq
+		{
+			CorrelationId = Uuid.NewUuid().ToDto(),
+			IsFinal = true,
+			Options = options
+		};
+		for (var index = 0; index < count; index++)
+		{
+			request.ProposedMessages.Add(new BatchAppendReq.Types.ProposedMessage
+			{
+				Id = Uuid.NewUuid().ToDto(),
+				Data = ByteString.CopyFromUtf8(data),
+				Metadata =
+				{
+					[GrpcMetadata.Type] = eventType,
+					[GrpcMetadata.ContentType] = GrpcMetadata.ContentTypes.ApplicationJson
+				}
+			});
+		}
+		await call.RequestStream.WriteAsync(request);
+		await call.RequestStream.CompleteAsync();
+		Assert.True(await call.ResponseStream.MoveNext());
+		return call.ResponseStream.Current;
+	}
+
+	public async Task<AppendResp> AppendSingle(string streamName, ulong? expectedRevision = null)
+	{
+		using var call = _client.Append(CallOptions);
+		var options = new AppendReq.Types.Options
+		{
+			StreamIdentifier = new() { StreamName = ByteString.CopyFromUtf8(streamName) }
+		};
+		if (expectedRevision.HasValue)
+			options.Revision = expectedRevision.Value;
+		else
+			options.Any = new();
+		await call.RequestStream.WriteAsync(new AppendReq
+		{
+			Options = options
+		});
+		await call.RequestStream.WriteAsync(new AppendReq
+		{
+			ProposedMessage = new()
+			{
+				Id = Uuid.NewUuid().ToDto(),
+				Data = ByteString.CopyFromUtf8("event"),
+				Metadata =
+				{
+					[GrpcMetadata.Type] = "event",
+					[GrpcMetadata.ContentType] = GrpcMetadata.ContentTypes.ApplicationJson
+				}
+			}
+		});
+		await call.RequestStream.CompleteAsync();
+		return await call.ResponseAsync;
+	}
+
+	public async Task<ReadResp[]> Read(
+		string streamName,
+		ulong revision,
+		ulong count,
+		ReadReq.Types.Options.Types.ReadDirection direction =
+			ReadReq.Types.Options.Types.ReadDirection.Forwards)
+	{
+		using var call = _client.Read(new ReadReq
+		{
+			Options = new()
+			{
+				Stream = new()
+				{
+					StreamIdentifier = new() { StreamName = ByteString.CopyFromUtf8(streamName) },
+					Revision = revision
+				},
+				Count = count,
+				ReadDirection = direction,
+				NoFilter = new(),
+				UuidOption = new() { Structured = new() }
+			}
+		}, CallOptions);
+		return await call.ResponseStream.ReadAllAsync().ToArrayAsync();
+	}
+
+	public void Dispose() => _channel?.Dispose();
+}

src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/HashCollisionGrpcBoundaryTests.cs

@@ -0,0 +1,89 @@
+using System;
+using System.IO;
+using System.Linq;
+using System.Threading.Tasks;
+using EventStore.Client.Streams;
+using EventStore.Core.Index;
+using EventStore.Core.Tests.Helpers;
+using NUnit.Framework;
+
+namespace EventStore.Core.Tests.Services.Transport.Grpc.StreamsTests;
+
+[Category("LongRunning")]
+public class HashCollisionGrpcBoundaryTests : SpecificationWithDirectoryPerTestFixture
+{
+	private const string FirstStream = "account--696193173";
+	private const string SecondStream = "LPN-FC002_LPK51001";
+	private MiniNode<LogFormat.V2, string> _node;
+	private GrpcStreamEdgeOperations _grpc;
+	private string _dbPath;
+
+	[OneTimeSetUp]
+	public override async Task TestFixtureSetUp()
+	{
+		await base.TestFixtureSetUp();
+		_dbPath = Path.Combine(PathName, "collision-node-db");
+		await StartNode(waitForAdminUserCreation: true);
+	}
+
+	[OneTimeTearDown]
+	public override async Task TestFixtureTearDown()
+	{
+		_grpc?.Dispose();
+		if (_node is not null)
+			await _node.Shutdown();
+		await base.TestFixtureTearDown();
+	}
+
+	[Test]
+	public async Task does_not_return_a_colliding_stream_after_the_read_limit_is_reached()
+	{
+		AssertSuccess(await _grpc.Append(FirstStream, noStream: true), 0);
+		AssertSuccess(await _grpc.Append(SecondStream, count: 100), 99);
+
+		await _node.Shutdown(keepDb: true);
+		_grpc.Dispose();
+		await StartNode(waitForAdminUserCreation: false);
+
+		var firstRead = await _grpc.Read(FirstStream, 0, 1);
+		Assert.That(firstRead.Single().ContentCase,
+			Is.EqualTo(ReadResp.ContentOneofCase.StreamNotFound));
+
+		var secondRead = await _grpc.Read(SecondStream, 99, 1);
+		Assert.That(secondRead.Single(x => x.Event is not null).Event.Event.StreamRevision,
+			Is.EqualTo(99));
+
+		var append = await _grpc.AppendSingle(FirstStream);
+		Assert.That(append.ResultCase, Is.EqualTo(AppendResp.ResultOneofCase.WrongExpectedVersion));
+		Assert.That(append.WrongExpectedVersion.CurrentRevisionOptionCase,
+			Is.EqualTo(AppendResp.Types.WrongExpectedVersion.CurrentRevisionOptionOneofCase.None));
+
+		var batchAppend = await _grpc.Append(FirstStream);
+		Assert.That(batchAppend.ResultCase, Is.EqualTo(BatchAppendResp.ResultOneofCase.Error));
+		Assert.That(batchAppend.Error.Code, Is.EqualTo(Google.Rpc.Code.AlreadyExists));
+		var error = batchAppend.Error.Details.Unpack<EventStore.Client.WrongExpectedVersion>();
+		Assert.That(error.CurrentStreamRevisionOptionCase,
+			Is.EqualTo(EventStore.Client.WrongExpectedVersion.CurrentStreamRevisionOptionOneofCase.None));
+	}
+
+	private async Task StartNode(bool waitForAdminUserCreation)
+	{
+		_node = new MiniNode<LogFormat.V2, string>(PathName,
+			dbPath: _dbPath,
+			memTableSize: 20,
+			hashCollisionReadLimit: 1,
+			indexBitnessVersion: PTableVersions.IndexV4,
+			hash32bit: true,
+			streamExistenceFilterSize: 0);
+		await _node.Start();
+		if (waitForAdminUserCreation)
+			await _node.AdminUserCreated;
+		_grpc = new GrpcStreamEdgeOperations(_node);
+	}
+
+	private static void AssertSuccess(BatchAppendResp response, ulong expectedRevision)
+	{
+		Assert.That(response.ResultCase, Is.EqualTo(BatchAppendResp.ResultOneofCase.Success));
+		Assert.That(response.Success.CurrentRevision, Is.EqualTo(expectedRevision));
+	}
+}