diff --git a/proto.lock b/proto.lock index 64536d3ba..ba29fa01c 100644 --- a/proto.lock +++ b/proto.lock @@ -6222,18 +6222,18 @@ "fields": [ { "id": 1, - "name": "stream_position", - "type": "uint64" + "name": "timestamp", + "type": "google.protobuf.Timestamp" }, { "id": 2, - "name": "all_stream_position", - "type": "event_store.client.AllStreamPosition" + "name": "stream_revision", + "type": "int64" }, { "id": 3, - "name": "no_position", - "type": "event_store.client.Empty" + "name": "position", + "type": "Position" } ] }, @@ -6242,18 +6242,18 @@ "fields": [ { "id": 1, - "name": "stream_position", - "type": "uint64" + "name": "timestamp", + "type": "google.protobuf.Timestamp" }, { "id": 2, - "name": "all_stream_position", - "type": "event_store.client.AllStreamPosition" + "name": "stream_revision", + "type": "int64" }, { "id": 3, - "name": "no_position", - "type": "event_store.client.Empty" + "name": "position", + "type": "Position" } ] }, @@ -6359,6 +6359,21 @@ } ] }, + { + "name": "Position", + "fields": [ + { + "id": 1, + "name": "commit_position", + "type": "uint64" + }, + { + "id": 2, + "name": "prepare_position", + "type": "uint64" + } + ] + }, { "name": "StreamNotFound", "fields": [ diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToAllTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToAllTests.cs index 9cb035adf..72067d648 100644 --- a/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToAllTests.cs +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToAllTests.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; @@ -19,6 +20,7 @@ public class when_subscribing_to_all : GrpcSpecification< private const string StreamId = nameof(when_subscribing_to_all); private readonly List _responses = new(); private Position _positionOfLastWrite; + private DateTime _subscriptionStartedAt; public when_subscribing_to_all() : base(new LotsOfExpiriesStrategy()) { @@ -43,6 +45,7 @@ protected override async Task Given() protected override async Task When() { + _subscriptionStartedAt = DateTime.UtcNow; using var call = StreamsClient.Read(new() { Options = new() @@ -83,8 +86,15 @@ public void subscription_confirmed() public void caught_up_includes_the_all_stream_position() { var caughtUp = _responses.Single(x => x.ContentCase == ReadResp.ContentOneofCase.CaughtUp).CaughtUp; - Assert.AreEqual(_positionOfLastWrite.CommitPosition, caughtUp.AllStreamPosition.CommitPosition); - Assert.AreEqual(_positionOfLastWrite.PreparePosition, caughtUp.AllStreamPosition.PreparePosition); + Assert.AreEqual(_positionOfLastWrite.CommitPosition, caughtUp.Position.CommitPosition); + Assert.AreEqual(_positionOfLastWrite.PreparePosition, caughtUp.Position.PreparePosition); + } + + [Test] + public void caught_up_includes_the_server_timestamp() + { + var caughtUp = _responses.Single(x => x.ContentCase == ReadResp.ContentOneofCase.CaughtUp).CaughtUp; + Assert.That(caughtUp.Timestamp.ToDateTime(), Is.InRange(_subscriptionStartedAt, DateTime.UtcNow)); } } diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToStreamTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToStreamTests.cs index a676b592b..64a3387f4 100644 --- a/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToStreamTests.cs +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToStreamTests.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; @@ -18,6 +19,7 @@ public class when_subscribing_to_stream : GrpcSpecificati private const string StreamId = nameof(when_subscribing_to_stream); private readonly List _responses = new(); private ulong _positionOfLastWrite; + private DateTime _subscriptionStartedAt; public when_subscribing_to_stream() : base(new LotsOfExpiriesStrategy()) { @@ -41,6 +43,7 @@ protected override async Task Given() protected override async Task When() { + _subscriptionStartedAt = DateTime.UtcNow; using var call = StreamsClient.Read(new() { Options = new() @@ -60,13 +63,9 @@ protected override async Task When() while (await call.ResponseStream.MoveNext()) { var response = call.ResponseStream.Current; - if (response.ContentCase == ReadResp.ContentOneofCase.Event && - _positionOfLastWrite == response.Event.Event.StreamRevision) - { - break; - } - _responses.Add(response); + if (response.ContentCase == ReadResp.ContentOneofCase.CaughtUp) + break; } // caught up, now add one more event @@ -93,5 +92,20 @@ public void subscription_confirmed() Assert.AreEqual(ReadResp.ContentOneofCase.Confirmation, _responses[0].ContentCase); Assert.NotNull(_responses[0].Confirmation.SubscriptionId); } + + [Test] + public void caught_up_includes_the_stream_revision() + { + var caughtUp = _responses.Single(x => x.ContentCase == ReadResp.ContentOneofCase.CaughtUp).CaughtUp; + Assert.True(caughtUp.HasStreamRevision); + Assert.AreEqual((long)_positionOfLastWrite, caughtUp.StreamRevision); + } + + [Test] + public void caught_up_includes_the_server_timestamp() + { + var caughtUp = _responses.Single(x => x.ContentCase == ReadResp.ContentOneofCase.CaughtUp).CaughtUp; + Assert.That(caughtUp.Timestamp.ToDateTime(), Is.InRange(_subscriptionStartedAt, DateTime.UtcNow)); + } } } diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscriptionStatusContractTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscriptionStatusContractTests.cs new file mode 100644 index 000000000..ab2b37004 --- /dev/null +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscriptionStatusContractTests.cs @@ -0,0 +1,36 @@ +using EventStore.Client.Streams; +using Google.Protobuf.Reflection; +using NUnit.Framework; + +namespace EventStore.Core.Tests.Services.Transport.Grpc.StreamsTests; + +[TestFixture] +public class SubscriptionStatusContractTests +{ + [Test] + public void caught_up_has_timestamp_and_optional_checkpoint_context() => + AssertStatusContract(ReadResp.Types.CaughtUp.Descriptor); + + [Test] + public void fell_behind_has_timestamp_and_optional_checkpoint_context() => + AssertStatusContract(ReadResp.Types.FellBehind.Descriptor); + + private static void AssertStatusContract(MessageDescriptor descriptor) + { + var timestamp = descriptor.FindFieldByNumber(1); + var streamRevision = descriptor.FindFieldByNumber(2); + var position = descriptor.FindFieldByNumber(3); + + Assert.Multiple(() => + { + Assert.That(timestamp.Name, Is.EqualTo("timestamp")); + Assert.That(timestamp.MessageType.FullName, Is.EqualTo("google.protobuf.Timestamp")); + Assert.That(streamRevision.Name, Is.EqualTo("stream_revision")); + Assert.That(streamRevision.FieldType, Is.EqualTo(FieldType.Int64)); + Assert.That(streamRevision.HasPresence, Is.True); + Assert.That(position.Name, Is.EqualTo("position")); + Assert.That(position.MessageType, Is.EqualTo(ReadResp.Types.Position.Descriptor)); + Assert.That(position.HasPresence, Is.True); + }); + } +} diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscriptionStatusMappingTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscriptionStatusMappingTests.cs new file mode 100644 index 000000000..8a37b9687 --- /dev/null +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscriptionStatusMappingTests.cs @@ -0,0 +1,39 @@ +using System; +using System.Reflection; +using EventStore.Core.Services.Transport.Enumerators; +using NUnit.Framework; + +namespace EventStore.Core.Tests.Services.Transport.Grpc.StreamsTests; + +[TestFixture] +public class SubscriptionStatusMappingTests +{ + private static readonly DateTime TransitionTimestamp = new(2026, 8, 24, 12, 34, 56, DateTimeKind.Utc); + + [Test] + public void caught_up_preserves_the_transition_timestamp() + { + var response = Map(new ReadResponse.SubscriptionCaughtUp(42, TransitionTimestamp)); + Assert.That(response.CaughtUp.Timestamp.ToDateTime(), Is.EqualTo(TransitionTimestamp)); + } + + [Test] + public void fell_behind_preserves_the_transition_timestamp() + { + var response = Map(new ReadResponse.SubscriptionFellBehind(42, TransitionTimestamp)); + Assert.That(response.FellBehind.Timestamp.ToDateTime(), Is.EqualTo(TransitionTimestamp)); + } + + private static EventStore.Client.Streams.ReadResp Map(ReadResponse response) + { + var streamsType = typeof(ReadResponse).Assembly + .GetType("EventStore.Core.Services.Transport.Grpc.Streams`1") + .MakeGenericType(typeof(string)); + var method = streamsType.GetMethod("TryConvertReadResponse", BindingFlags.NonPublic | BindingFlags.Static); + var arguments = new object[] { response, null, null }; + + Assert.That(method, Is.Not.Null); + Assert.That(method.Invoke(null, arguments), Is.True); + return (EventStore.Client.Streams.ReadResp)arguments[2]; + } +} diff --git a/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscription.cs b/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscription.cs index 782715508..dc64da5bf 100644 --- a/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscription.cs +++ b/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscription.cs @@ -202,7 +202,7 @@ private async Task NotifyCaughtUp(TFPos checkpoint, CancellationToken ct) "Subscription {subscriptionId} to $all caught up at checkpoint {position}.", _subscriptionId, checkpoint); - await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionCaughtUp(checkpoint), ct); + await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionCaughtUp(checkpoint, DateTime.UtcNow), ct); } private async Task NotifyFellBehind(TFPos checkpoint, CancellationToken ct) @@ -211,7 +211,7 @@ private async Task NotifyFellBehind(TFPos checkpoint, CancellationToken ct) "Subscription {subscriptionId} to $all fell behind at checkpoint {position}.", _subscriptionId, checkpoint); - await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionFellBehind(checkpoint), ct); + await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionFellBehind(checkpoint, DateTime.UtcNow), ct); } private async ValueTask<(TFPos, ulong)> GoLive(TFPos checkpoint, ulong sequenceNumber, CancellationToken ct) diff --git a/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscriptionFiltered.cs b/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscriptionFiltered.cs index 5bb1b23bf..720062cbd 100644 --- a/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscriptionFiltered.cs +++ b/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscriptionFiltered.cs @@ -235,7 +235,7 @@ private async Task NotifyCaughtUp(TFPos checkpoint, CancellationToken ct) "Subscription {subscriptionId} to $all:{eventFilter} caught up at checkpoint {position}.", _subscriptionId, _eventFilter, checkpoint); - await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionCaughtUp(checkpoint), ct); + await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionCaughtUp(checkpoint, DateTime.UtcNow), ct); } private async Task NotifyFellBehind(TFPos checkpoint, CancellationToken ct) @@ -244,7 +244,7 @@ private async Task NotifyFellBehind(TFPos checkpoint, CancellationToken ct) "Subscription {subscriptionId} to $all:{eventFilter} fell behind at checkpoint {position}.", _subscriptionId, _eventFilter, checkpoint); - await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionFellBehind(checkpoint), ct); + await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionFellBehind(checkpoint, DateTime.UtcNow), ct); } private async ValueTask<(TFPos, ulong)> GoLive(TFPos checkpoint, ulong sequenceNumber, CancellationToken ct) diff --git a/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.StreamSubscription.cs b/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.StreamSubscription.cs index 3162da133..b58380a9d 100644 --- a/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.StreamSubscription.cs +++ b/src/EventStore.Core/Services/Transport/Enumerators/Enumerator.StreamSubscription.cs @@ -218,7 +218,7 @@ private async Task NotifyCaughtUp(long checkpoint, CancellationToken ct) "Subscription {subscriptionId} to {streamName} caught up at checkpoint {streamRevision:N0}.", _subscriptionId, _streamName, checkpoint); - await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionCaughtUp(checkpoint), ct); + await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionCaughtUp(checkpoint, DateTime.UtcNow), ct); } private async Task NotifyFellBehind(long checkpoint, CancellationToken ct) @@ -227,7 +227,7 @@ private async Task NotifyFellBehind(long checkpoint, CancellationToken ct) "Subscription {subscriptionId} to {streamName} fell behind at checkpoint {streamRevision:N0}.", _subscriptionId, _streamName, checkpoint); - await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionFellBehind(checkpoint), ct); + await _channel.Writer.WriteAsync(new ReadResponse.SubscriptionFellBehind(checkpoint, DateTime.UtcNow), ct); } private async ValueTask<(long, ulong)> GoLive(long checkpoint, ulong sequenceNumber, CancellationToken ct) diff --git a/src/EventStore.Core/Services/Transport/Enumerators/ReadResponse.cs b/src/EventStore.Core/Services/Transport/Enumerators/ReadResponse.cs index 52a3fdd6a..a23cf052a 100644 --- a/src/EventStore.Core/Services/Transport/Enumerators/ReadResponse.cs +++ b/src/EventStore.Core/Services/Transport/Enumerators/ReadResponse.cs @@ -1,3 +1,4 @@ +using System; using EventStore.Core.Data; using EventStore.Core.Services.Transport.Common; @@ -17,33 +18,39 @@ public EventReceived(ResolvedEvent @event) public class SubscriptionCaughtUp : ReadResponse { + public readonly DateTime Timestamp; public readonly TFPos? AllStreamPosition; public readonly long? StreamPosition; - public SubscriptionCaughtUp(TFPos allStreamPosition) + public SubscriptionCaughtUp(TFPos allStreamPosition, DateTime timestamp) { AllStreamPosition = allStreamPosition; + Timestamp = timestamp; } - public SubscriptionCaughtUp(long streamPosition) + public SubscriptionCaughtUp(long streamPosition, DateTime timestamp) { StreamPosition = streamPosition; + Timestamp = timestamp; } } public class SubscriptionFellBehind : ReadResponse { + public readonly DateTime Timestamp; public readonly TFPos? AllStreamPosition; public readonly long? StreamPosition; - public SubscriptionFellBehind(TFPos allStreamPosition) + public SubscriptionFellBehind(TFPos allStreamPosition, DateTime timestamp) { AllStreamPosition = allStreamPosition; + Timestamp = timestamp; } - public SubscriptionFellBehind(long streamPosition) + public SubscriptionFellBehind(long streamPosition, DateTime timestamp) { StreamPosition = streamPosition; + Timestamp = timestamp; } } diff --git a/src/EventStore.Core/Services/Transport/Grpc/Streams.Read.cs b/src/EventStore.Core/Services/Transport/Grpc/Streams.Read.cs index 703d66464..e00c2c333 100644 --- a/src/EventStore.Core/Services/Transport/Grpc/Streams.Read.cs +++ b/src/EventStore.Core/Services/Transport/Grpc/Streams.Read.cs @@ -17,6 +17,7 @@ using FilterOptionOneofCase = EventStore.Client.Streams.ReadReq.Types.Options.FilterOptionOneofCase; using ReadDirection = EventStore.Client.Streams.ReadReq.Types.Options.Types.ReadDirection; using StreamOptionOneofCase = EventStore.Client.Streams.ReadReq.Types.Options.StreamOptionOneofCase; +using Timestamp = Google.Protobuf.WellKnownTypes.Timestamp; namespace EventStore.Core.Services.Transport.Grpc; @@ -314,56 +315,58 @@ private static bool TryConvertReadResponse(ReadResponse readResponse, ReadReq.Ty private static ReadResp.Types.CaughtUp ToCaughtUp(ReadResponse.SubscriptionCaughtUp caughtUp) { - var response = new ReadResp.Types.CaughtUp(); + var response = new ReadResp.Types.CaughtUp + { + Timestamp = Timestamp.FromDateTime(caughtUp.Timestamp) + }; SetPosition(response, caughtUp.AllStreamPosition, caughtUp.StreamPosition); return response; } private static ReadResp.Types.FellBehind ToFellBehind(ReadResponse.SubscriptionFellBehind fellBehind) { - var response = new ReadResp.Types.FellBehind(); + var response = new ReadResp.Types.FellBehind + { + Timestamp = Timestamp.FromDateTime(fellBehind.Timestamp) + }; SetPosition(response, fellBehind.AllStreamPosition, fellBehind.StreamPosition); return response; } private static void SetPosition(ReadResp.Types.CaughtUp response, TFPos? allStreamPosition, long? streamPosition) { - if (TrySetAllStreamPosition(allStreamPosition, position => response.AllStreamPosition = position)) + if (TrySetAllStreamPosition(allStreamPosition, position => response.Position = position)) { return; } - if (TrySetStreamPosition(streamPosition, position => response.StreamPosition = position)) + if (TrySetStreamPosition(streamPosition, position => response.StreamRevision = position)) { return; } - - response.NoPosition = new Empty(); } private static void SetPosition(ReadResp.Types.FellBehind response, TFPos? allStreamPosition, long? streamPosition) { - if (TrySetAllStreamPosition(allStreamPosition, position => response.AllStreamPosition = position)) + if (TrySetAllStreamPosition(allStreamPosition, position => response.Position = position)) { return; } - if (TrySetStreamPosition(streamPosition, position => response.StreamPosition = position)) + if (TrySetStreamPosition(streamPosition, position => response.StreamRevision = position)) { return; } - - response.NoPosition = new Empty(); } - private static bool TrySetAllStreamPosition(TFPos? position, Action setPosition) + private static bool TrySetAllStreamPosition(TFPos? position, Action setPosition) { if (position is not { CommitPosition: >= 0, PreparePosition: >= 0 } allPosition) { return false; } - setPosition(new AllStreamPosition + setPosition(new ReadResp.Types.Position { CommitPosition = (ulong)allPosition.CommitPosition, PreparePosition = (ulong)allPosition.PreparePosition @@ -372,14 +375,14 @@ private static bool TrySetAllStreamPosition(TFPos? position, Action setPosition) + private static bool TrySetStreamPosition(long? streamPosition, Action setPosition) { if (streamPosition is not >= 0) { return false; } - setPosition((ulong)streamPosition); + setPosition(streamPosition.Value); return true; } diff --git a/src/Protos/Grpc/streams.proto b/src/Protos/Grpc/streams.proto index 1212e9bff..e3911675a 100644 --- a/src/Protos/Grpc/streams.proto +++ b/src/Protos/Grpc/streams.proto @@ -104,19 +104,15 @@ message ReadResp { } message CaughtUp { - oneof position { - uint64 stream_position = 1; - event_store.client.AllStreamPosition all_stream_position = 2; - event_store.client.Empty no_position = 3; - } + google.protobuf.Timestamp timestamp = 1; + optional int64 stream_revision = 2; + optional Position position = 3; } message FellBehind { - oneof position { - uint64 stream_position = 1; - event_store.client.AllStreamPosition all_stream_position = 2; - event_store.client.Empty no_position = 3; - } + google.protobuf.Timestamp timestamp = 1; + optional int64 stream_revision = 2; + optional Position position = 3; } message ReadEvent { @@ -145,6 +141,10 @@ message ReadResp { uint64 commit_position = 1; uint64 prepare_position = 2; } + message Position { + uint64 commit_position = 1; + uint64 prepare_position = 2; + } message StreamNotFound { event_store.client.StreamIdentifier stream_identifier = 1; }