How to disable certain EventHandlers when restoring from EventStore

[iappwebdev]

Hey,

I am using EventStoreDB for storing my events and MongoDb for projections/read models. I have two types of Subscriptions:

1. EventHandlers building read models

Example 1

On<BellRinged>(
    builder => builder
        .UpdateOne
        .DefaultId()
        .UpdateFromContext(
            (ctx, update) => update
                .Set(x => x.HasBellRinged, true)
        )
);

2. EventHandlers performing new Commands to other aggregates or executing other infrastructure stuff, e.g. sending notifications

Example 2

On<BellRinged>(async ctx => await _notificationService.Handle(
        new NotifyAdminThatBellRinged(),
        CancellationToken.None
    )
    .ConfigureAwait(true)
);

I works fine until I want to rebuild my read models and delete the corresponding collections and the checkpoints.

When I now emit a new event, the read models get rebuild (1). That's fine.
But the EventHandlers performing new Command to other Aggregates are executed as well (2), that is not fine, these events shouldn't be executed again.

How can I disable these events or the whole eventhandlers. Do I have to save somewhere the messageId and perfom manual checks?
Or are there other options.

Thanks for help and answers!

[iappwebdev]

I did some research and came up with two possible solutions(?).

1. Attempt

I created a custom OrchestratorHandler inheriting from EventHandler. I override HandleEvent so that passed event are ignored, when their Created-Date is before the instantiation of the event handler.

public abstract class OrchestratorHandler  : EventHandler
{
    private readonly DateTime _subscriptionsFrom;

    protected OrchestratorHandler(TypeMapper? typeMap = null)
        : base(typeMap) =>
        _subscriptionsFrom = DateTime.Now;

    public override async ValueTask<EventHandlingStatus> HandleEvent(IMessageConsumeContext context)
    {
        Guard.Against.Null(context);
        if (context.Created.ToUniversalTime() < _subscriptionsFrom.ToUniversalTime())
        {
            return EventHandlingStatus.Ignored;
        }

        return await base.HandleEvent(context).ConfigureAwait(false);
    }
}

This works, but has the drawback, that the event handler gets all events from the beginning when the checkpoint is deleted.

2. Attempt

I thought again about this problem, had a look in the EventStoreDB docs and realized that these are two types of event handers: catch-up and live subscriptions. Catch-up subscriptions can subscribe from the start or from a specific position. Live subscriptions subscribe from the end of the stream.

Subscribing from the start
Subscribing from a specific position
Subscribing to a stream for live updates

In Eventuous, I use AllStreamSubscription, inherit from EventStoreCatchUpSubscriptionBase (namespace Eventuous.EventStore.Subscriptions).

/// <summary>
/// Catch-up subscription for EventStoreDB, using the $all global stream
/// </summary>
[PublicAPI]
public class AllStreamSubscription
    : EventStoreCatchUpSubscriptionBase<AllStreamSubscriptionOptions>, IMeasuredSubscription {
    /// <summary>
    /// Creates EventStoreDB catch-up subscription service for $all
    /// </summary>
    /// <param name="eventStoreClient">EventStoreDB gRPC client instance</param>
    /// <param name="subscriptionId">Subscription ID</param>
    /// <param name="checkpointStore">Checkpoint store instance</param>
    /// <param name="consumePipe"></param>
    /// <param name="eventSerializer">Event serializer instance</param>
    /// <param name="metaSerializer"></param>
    /// <param name="eventFilter">Optional: server-side event filter</param>
    /// <param name="loggerFactory"></param>

In there we have

protected override async ValueTask Subscribe(CancellationToken cancellationToken) {
        var filterOptions = new SubscriptionFilterOptions(
            Options.EventFilter ?? EventTypeFilter.ExcludeSystemEvents(),
            Options.CheckpointInterval
        );

        var (_, position) = await GetCheckpoint(cancellationToken).NoContext();

        var fromAll = position == null ? FromAll.Start : FromAll.After(new Position(position.Value, position.Value));

        Subscription = await EventStoreClient.SubscribeToAllAsync(
                fromAll,
                (_, @event, ct) => HandleEvent(@event, ct),
                Options.ResolveLinkTos,
                HandleDrop,
                filterOptions,
                Options.Credentials,
                cancellationToken
            )
            .NoContext();

In there the position is defined to start from start or from the last known position. So I need to set the position to FromAll.End. But there is no way to modify this, respectively I could not find a way. So I inherited from AllStreamSubscription and overrode ValueTask Subscribe(CancellationToken cancellationToken) and set the to FromAll.End:

public sealed class AllStreamLiveSubscription : AllStreamSubscription
{
    public AllStreamLiveSubscription(EventStoreClient eventStoreClient, string subscriptionId, ICheckpointStore checkpointStore, ConsumePipe consumePipe, IEventSerializer? eventSerializer = null,
        IMetadataSerializer? metaSerializer = null,
        IEventFilter? eventFilter = null,
        ILoggerFactory? loggerFactory = null)
        : base(eventStoreClient, subscriptionId, checkpointStore, consumePipe, eventSerializer, metaSerializer, eventFilter, loggerFactory)
    { }

    public AllStreamLiveSubscription(EventStoreClient eventStoreClient, AllStreamSubscriptionOptions options, ICheckpointStore checkpointStore, ConsumePipe consumePipe, ILoggerFactory? loggerFactory)
        : base(eventStoreClient, options, checkpointStore, consumePipe, loggerFactory)
    { }

    protected override async ValueTask Subscribe(CancellationToken cancellationToken) {
        var filterOptions = new SubscriptionFilterOptions(
            Options.EventFilter ?? EventTypeFilter.ExcludeSystemEvents(),
            Options.CheckpointInterval
        );

        Subscription = await EventStoreClient.SubscribeToAllAsync(
                FromAll.End,
                (_, @event, ct) => HandleEvent(@event, ct),
                Options.ResolveLinkTos,
                HandleDrop,
                filterOptions,
                Options.Credentials,
                cancellationToken
            )
            .ConfigureAwait(false);

        async Task HandleEvent(ResolvedEvent re, CancellationToken ct)
            => await HandleInternal(CreateContext(re, ct)).ConfigureAwait(false);

        void HandleDrop(global::EventStore.Client.StreamSubscription _, SubscriptionDroppedReason reason, Exception? ex)
            => Dropped(EsdbMappings.AsDropReason(reason), ex);
    }

    IMessageConsumeContext CreateContext(ResolvedEvent re, CancellationToken cancellationToken) {
        var evt = DeserializeData(
            re.Event.ContentType,
            re.Event.EventType,
            re.Event.Data,
            re.Event.EventStreamId,
            re.Event.EventNumber
        );

        return new MessageConsumeContext(
            re.Event.EventId.ToString(),
            re.Event.EventType,
            re.Event.ContentType,
            re.Event.EventStreamId,
            re.Event.EventNumber,
            re.OriginalEventNumber,
            re.Event.Position.CommitPosition,
            _sequence++,
            re.Event.Created,
            evt,
            Options.MetadataSerializer.DeserializeMeta(Options, re.Event.Metadata, re.Event.EventStreamId),
            SubscriptionId,
            cancellationToken
        );
    }

    ulong _sequence;
}

static class EsdbMappings {
    public static DropReason AsDropReason(SubscriptionDroppedReason reason)
        => reason switch {
            SubscriptionDroppedReason.Disposed => DropReason.Stopped,
            SubscriptionDroppedReason.ServerError => DropReason.ServerError,
            SubscriptionDroppedReason.SubscriberError => DropReason.SubscriptionError,
            _ => throw new ArgumentOutOfRangeException(nameof(reason), reason, null)
        };

}

In my Program.cs:

services.AddSubscription<AllStreamLiveSubscription, AllStreamSubscriptionOptions>(
    "MyEventHandlerStreamName",
    builder => builder
        .UseCheckpointStore<NoOpCheckpointStore>()
        .AddEventHandler<MyEventHandler>()
);
...
static SubscriptionBuilder<AppLiveAllStreamSubscription, AllStreamSubscriptionOptions> UseCheckpointStore<T>(
        this SubscriptionBuilder<AppLiveAllStreamSubscription, AllStreamSubscriptionOptions> builder
    ) where T : class, ICheckpointStore
        => builder.UseCheckpointStore<AppLiveAllStreamSubscription, AllStreamSubscriptionOptions, T>();

Two remarks: I could not use MongoCheckpointStore anymore because I got this error

fail: Eventuous.Subscription[0]
      [MyEventHandler] Unable to commit position CommitPosition { Position: 18077, Sequence: 1, Timestamp: 2023-12-01T07:41:42.1731952Z }
      System.Collections.Generic.KeyNotFoundException: The given key 'MyEventHandlerStreamName' was not present in the dictionary.
         at System.Collections.Generic.Dictionary`2.get_Item(TKey key)
         at Eventuous.Projections.MongoDB.MongoCheckpointStore.StoreCheckpoint(Checkpoint checkpoint, Boolean force, CancellationToken cancellationToken)
         at Eventuous.Subscriptions.Checkpoints.MeasuredCheckpointStore.StoreCheckpoint(Checkpoint checkpoint, Boolean force, CancellationToken cancellationToken)
         at Eventuous.Subscriptions.Checkpoints.CheckpointCommitHandler.CommitInternal(CommitPosition position, Boolean force, CancellationToken cancellationToken)

So I changed it to NoOpCheckpointStore because I think that I do need checkpoint in this case. Am I right, what do you think about this approach. Are there any drawbacks or things to improve?

What I also didn't like was to be forced to copy a lot of code out of AllStreamSubscription, maybe there is a better solution?

[alexeyzimarev]

For the checkpoint store problem, maybe you can open an issue. I cannot see immediately what happens. Checkpointing is being tested, and I haven't seen anything like that. MongoDB module tests use MongoDB checkpoint store, and everything seems to work.

[alexeyzimarev]

There are two things I always suggest:

• When using catch-up subscriptions for reactions (your handler that triggers NotifyAdminThatBellRinged) must be separated from the subscription that produces read models. It is unconditional, and applies to the next point as well.
• Reactions should never be replayed, as the previous point suggests. So, storing checkpoints in a database that works great with read models becomes less meaningful. Therefore, when using ESDB I would suggest using a persistent subscription instead as it uses the server-managed checkpointing.

[alexeyzimarev]

I could also suggest using a persistent subscription for things like triggering notifications. By doing so, you'd decrease the chance of your checkpoint being lost, as ESDB will take care of it.