Skip to content

Subscription gap metrics: GetSubscriptionEndOfStream issues #548

@wpavelev

Description

@wpavelev

Describe the bug

We are encountering periodic warning log entries with the message "Failed to get end of stream" in applications using SQL-Server-backed Eventuous subscriptions. After investigation, SqlSubscriptionBase.GetSubscriptionEndOfStream has three bugs, all causing the periodic "Failed to get end of stream" warning:

  1. Swapped switch arms: SubscriptionKind.All executes GetEndOfStream (stream-scoped MAX(stream_position)) instead of GetEndOfAll (global MAX(global_position)), and vice versa.
  2. Type mismatch on non-empty tables: reader.GetInt64(0) is used to read the result of MAX(stream_position) / MAX(global_position). ADO.NET's typed getter is strict — if the database driver returns the column as int (32-bit), GetInt64 throws an InvalidCastException. This happens even when events are present.
  3. No DBNull guard on empty tables: MAX() returns NULL when the table is empty, causing GetInt64(0) to throw InvalidCastException regardless of the type issue above.

The warning appears periodically because GetSubscriptionEndOfStream is exposed via IMeasuredSubscription.GetMeasure() and called repeatedly by the subscription diagnostics/metrics system to measure subscription lag.

Additional context

The fix in GetSubscriptionEndOfStream:

async ValueTask<EndOfStream> GetSubscriptionEndOfStream(CancellationToken cancellationToken) {
        try {
            await using var connection = await OpenConnection(cancellationToken).NoContext();
            await using var cmd        = connection.CreateCommand();
            cmd.CommandType = CommandType.Text;

            cmd.CommandText = Kind switch {
                SubscriptionKind.All    => GetEndOfAll,
                SubscriptionKind.Stream => GetEndOfStream
            };
            await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).NoContext();

            var position = await reader.ReadAsync(cancellationToken).NoContext() && reader[0] is not DBNull
                ? Convert.ToInt64(reader[0])
                : 0;

            return new(SubscriptionId, (ulong)position, DateTime.UtcNow);
        } catch (Exception e) {
            Log.WarnLog?.Log(e, "Failed to get end of stream");

            return EndOfStream.Invalid;
        }
}

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type
    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions