Skip to content

feat(streams): expose subscription status context #463

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yordis merged 1 commit into from
Aug 24, 2026
Merged

feat(streams): expose subscription status context #463

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 27 additions & 12 deletions proto.lock
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
}
]
},
Expand All @@ -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"
}
]
},
Expand Down Expand Up @@ -6359,6 +6359,21 @@
}
]
},
{
"name": "Position",
"fields": [
{
"id": 1,
"name": "commit_position",
"type": "uint64"
},
{
"id": 2,
"name": "prepare_position",
"type": "uint64"
}
]
},
{
"name": "StreamNotFound",
"fields": [
Expand Down
14 changes: 12 additions & 2 deletions src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToAllTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
Expand All @@ -19,6 +20,7 @@ public class when_subscribing_to_all<TLogFormat, TStreamId> : GrpcSpecification<
private const string StreamId = nameof(when_subscribing_to_all<TLogFormat, TStreamId>);
private readonly List<ReadResp> _responses = new();
private Position _positionOfLastWrite;
private DateTime _subscriptionStartedAt;

public when_subscribing_to_all() : base(new LotsOfExpiriesStrategy())
{
Expand All @@ -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()
Expand Down Expand Up @@ -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));
}
}

Expand Down
26 changes: 20 additions & 6 deletions src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscribeToStreamTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
Expand All @@ -18,6 +19,7 @@ public class when_subscribing_to_stream<TLogFormat, TStreamId> : GrpcSpecificati
private const string StreamId = nameof(when_subscribing_to_stream<TLogFormat, TStreamId>);
private readonly List<ReadResp> _responses = new();
private ulong _positionOfLastWrite;
private DateTime _subscriptionStartedAt;

public when_subscribing_to_stream() : base(new LotsOfExpiriesStrategy())
{
Expand All @@ -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()
Expand All @@ -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
Expand All @@ -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));
}
}
}
36 changes: 36 additions & 0 deletions ...tStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscriptionStatusContractTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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);
});
}
}
39 changes: 39 additions & 0 deletions ...ntStore.Core.Tests/Services/Transport/Grpc/StreamsTests/SubscriptionStatusMappingTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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];
}
}
4 changes: 2 additions & 2 deletions src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscription.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand Down
4 changes: 2 additions & 2 deletions src/EventStore.Core/Services/Transport/Enumerators/Enumerator.AllSubscriptionFiltered.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand Down
4 changes: 2 additions & 2 deletions src/EventStore.Core/Services/Transport/Enumerators/Enumerator.StreamSubscription.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand Down
15 changes: 11 additions & 4 deletions src/EventStore.Core/Services/Transport/Enumerators/ReadResponse.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
using System;
using EventStore.Core.Data;
using EventStore.Core.Services.Transport.Common;

Expand All @@ -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;
}
}

Expand Down
Loading
Loading