Retroactive Event

Example: Retroactive Cargoes with Rewind (C#)

I've been illustrating most of the examples around Event Sourcing with an example based on moving cargoes on ships between ports. I shall continue with that example, looking this time on how to use Retroactive Event.

First off we need to make some of the strategic decisions around how to do our Retroactive Event. For this example I'll use the approach of reaching the branch point by rewinding the live model. Since the example is simple I won't use any selective replay, instead I'll just rewind the full event list.

Since we are using rewind it's essential that all of the events are reversible. I won't go into the details of the event processing or reversing, see the examples in Event Sourcing to see how that might work.

The Retroactive Event behavior is triggered by a special kind of event.

class ReplacementEvent...

  private DomainEvent original, replacement;
  public DomainEvent Original {get { return original; }}
  public DomainEvent Replacement {get { return replacement; }}

  public ReplacementEvent(DomainEvent oldEvent, DomainEvent replacement)
    : base(oldEvent.Occurred) {
    this.original = oldEvent;
    this.replacement = replacement;
  }
  internal override void Process() {
    throw new Exception("Replacements should not be processed directly");
  }
  internal override void Reverse() {
    throw new Exception("Cannot reverse replacements");
  }

You'll notice that in these cases I've blocked the basic process and reverse methods. Most of the time in these examples I've preferred to have the events handle their own processing logic. Replacements are different: they don't contain any domain logic, and their processing involves knowledge of the event queue and manipulating that event queue. Intimate interaction with the event queue is something I'd prefer to leave to the event processor alone, so in fact I built all the replacement behavior into the event queue.

class EventProcessor...

  public void Process(DomainEvent ev) {
    try {
      if (ev is ReplacementEvent)
        ProcessReplacement((ReplacementEvent) ev);
      else if (OutOfOrder(ev))
        ProcessOutOfOrder(ev);
      else BasicProcessEvent(ev);
    } catch (Exception ex) {
      ev.ProcessingError = ex;
      if (ShouldRethrowExceptions) throw ex;
    }
    InsertToLog(ev);
  }

You'll notice here that I'm committing the cardinal sin of object-oriented programming: explicit conditional behavior based on the type of the argument to a method. I don't do this often, but here I do it because I want the event processor to keep knowledge of the queue to itself. (I could also use double dispatch, but here the situation doesn't seem to be complex enough to warrant it.)

You may also notice that I don't have a case for rejected events here. As we'll see, I deal with those by treating those as replacements where the replacing event is null.

I'll start with the simplest case. The basic Process (and reverse) method is just a simple wrapper that allows me throw in some tracing behavior if I need it.

class EventProcessor...

  private void BasicProcessEvent(DomainEvent e) {
    traceProcess(e);
    e.Process();
  }
  private void BasicReverseEvent(DomainEvent e) {
    traceReverse(e);
    e.Reverse();
  }

Out of order events are the easiest to describe first. These are just regular events that we received out of sequence, we don't use replacement events to model them.Essentially these insert a new event into the stream. The processor tests for an out of order event by simply comparing it to the last one.

class EventProcessor...

  private bool OutOfOrder(DomainEvent e) {
    if (LogIsEmpty()) return false;
    return (LastEvent().after(e));
  }
  private DomainEvent LastEvent() {
    if (LogIsEmpty()) return null;
    return (DomainEvent) log[log.Count - 1];
  }
  private bool LogIsEmpty() {
    return 0 == log.Count;
  }

To keep things laughably simple for this example, I just do event ordering by the occurred date. To do this for real, it may be sufficient just to go to a finer resolution of time point. Sometimes other factors may affect orderings.

The outline of processing out order cases is simple.

class EventProcessor...

  private void ProcessOutOfOrder(DomainEvent e) {
    RewindTo(e);
    BasicProcessEvent(e);
    ReplayAfter(e);
  }

To rewind, I just select all the events from the log that are later than the out of order event. Again for simplicity I'm keeping my log (or at least a cache of it) in memory.

class EventProcessor...

  private void RewindTo(DomainEvent priorEvent) {
    IList consequences = Consequences(priorEvent);
    for (int i = consequences.Count - 1; i >= 0; i--)
      BasicReverseEvent(((DomainEvent) consequences[i]));
  }
  private IList Consequences(DomainEvent baseEvent) {
    IList result = new ArrayList();
    foreach (DomainEvent candidate in log)
      if (candidate.IsConsequenceOf(baseEvent)) result.Add(candidate);
    return result;
  }

class DomainEvent...

  public bool IsConsequenceOf(DomainEvent other) {
    return (!ShouldIgnoreOnReplay && this.after(other));
  }

As you can see, not all events are selected for reprocessing during a rewind, even though I'm not using any selective replay. Essentially I'm not reprocessing either error events or events that have been rejected.

Replaying events forwards again is simple

class EventProcessor...

  private void ReplayAfter(DomainEvent ev) {
    foreach (DomainEvent e in Consequences(ev)) BasicProcessEvent(e);
  }

Now lets move onto replacements.

class EventProcessor...

  private void ProcessReplacement(ReplacementEvent e) {
    if (e.Original.ShouldIgnoreOnReplay)
      throw new ProcessingException("Cannot replace event twice");
    else if (null == e.Replacement) 
      ProcessRejection(e);
    else if (e.HasPriorReplacement) 
      ProcessPriorReplacement(e);
    else 
      ProcessPriorOriginal(e);
    
  }

class ReplacementEvent...

  public bool HasPriorReplacement {
    get {
      if (null == replacement) return false;
      else return original.after(replacement);
    }
  }

There are several cases to process here. If the original event is already marked to ignore, that means there's an processing error as we should not be rejecting events that are already rejected. Once we've done that we have three main cases: the replacement is a rejection, the replacement event is earlier than the original, and the original is earlier than the replacement.

Let's start with the rejection case, which is indicated by a null replacement event. Here we rewind to the rejected event, reject it, reverse it, and replay forwards.

class EventProcessor...

  private void ProcessRejection(ReplacementEvent e) {
    RewindTo(e.Original);
    BasicReverseEvent(e.Original);
    e.Original.Reject();
    ReplayAfter(e.Original);
  }

class DomainEvent...

  public bool after (DomainEvent other) {
    return this.CompareTo(other) > 0;
  }
  public void Reject() {
    _isRejected = true;
  }
  private bool _isRejected;
  public virtual bool ShouldIgnoreOnReplay {
    get {
      if (WasProcessingError) return true;
      return _isRejected;
    }
  }

Rejecting marks the event so it won't get processed or rewound again.

If we have a prior replacement, we rewind to the replacement, reject the original, process the replacement and replay forwards.

class EventProcessor...

  private void ProcessPriorReplacement(ReplacementEvent e) {
    RewindTo(e.Replacement);
    e.Original.Reject();
    BasicProcessEvent(e.Replacement);
    ReplayAfter(e.Replacement);
  }

For a prior original, we rewind to the original, reverse and reject it, replay to the replacement, process it, and continue replay forwards.

class EventProcessor...

  private void ProcessPriorOriginal(ReplacementEvent e) {
    RewindTo(e.Original);
    BasicReverseEvent(e.Original);
    e.Original.Reject();
    ReplayBetween(e.Original, e.Replacement);
    BasicProcessEvent(e.Replacement);
    ReplayAfter(e.Replacement);
  }
  private void ReplayBetween(DomainEvent first, DomainEvent last) {
    IList eventsToReplay = new ArrayList();
    foreach (DomainEvent e in log) {
      if (e.IsConsequenceOf(first) && last.after(e))
        eventsToReplay.Add(e);
    }
    foreach (DomainEvent e in eventsToReplay)
      BasicProcessEvent(e);
  }

I'm following a convention here of not adding an event to the log until it's been successfully processed. This is useful if the log is not in the same transaction as the application state. If it's all in one transaction I could just add the event into the log after I've reached the branch point and replay forwards (which would pick up the new event.)

Example: Updates to External Systems (C#)

For simple Event Sourcing, it suffices to switch off the external notification during replay. For Retroactive Event, however, we need to go further and do detection and correction. To do this during the rewind/replay we'll save all the events generated and compare to determine what we need to correct. We'll then assume a simple automated correction case of sending cancellation messages and new messages for the changes.

I'll continue with the shipping example here by saying the the customs authorities in the US need to be notified should any cargo enter a US port that has passed through Canada.

The external notification is made by the Cargo object when it's handling an arrival event.

class Cargo...

  public void HandleArrival(ArrivalEvent ev) {
    ev.priorCargoInCanada[this] = _hasBeenInCanada;
    if ("CA" == ev.Port.Country) 
      _hasBeenInCanada = true;
    if (HasBeenInCanada && "US" == ev.Port.Country) {
      Registry.CustomsGateway.Notify(ev.Occurred, ev.Ship, ev.Port);
      ev.WasNotificationSent = true;
    }
  }
  private bool _hasBeenInCanada = false;
  public bool HasBeenInCanada {get { return _hasBeenInCanada;}}

To reverse this event we keep a note of whether we sent a notification in the forward play and send it again when we did.

class Cargo...

  public void ReverseArrival(ArrivalEvent ev) {
    _hasBeenInCanada = (bool) ev.priorCargoInCanada[this];
    if (ev.WasNotificationSent) 
      Registry.CustomsGateway.Notify(ev.Occurred, ev.Ship, ev.Port);
  }

I'm following the principle here that the domain model should be ignorant of the replay logic in the event processing. It knows how to reverse its own state for each event, but doesn't care about the complexities of talking to external systems.

The logic of talking to external systems is handled by the gateway, which in this case is a cluster of objects.

Figure 1: A cluster of objects make up the gateway to handle retro-activity.

In this example the CustomsGatewayFront is a normal gateway that converts a domain-oriented interface (defined by ICustomsGateway) to the actual messaging infrastructure. We can safely ignore what it does - we just assume that if we invoke a method on it, it ensures that the customs office gets the message.

The interesting stuff happens before, driven by CustomsGatewayBuffer which wraps the actual CustomsGatewayFront, implementing the same interface but adding the ability to cope with the retroactive rewinding.

Because much of the retroactive rewinding is generic, I can put the generic behavior into a different class, ReplayBuffer, so that the Customs Gateway Buffer only handles stuff that's particular to the customs case. The replay buffer does need some communication with the event processor to deal with various stages in the rewind/replay process. It also needs some way to adjust the mismatching events in the end, a task that in this case I assign to the customs gateway buffer. In both of these cases the rewind buffer communicates through collaboration interfaces in order to keep its generic purity.

So thats the dramatis personae, lets now go into the action. I'll begin with the normal case when we are running live. The customs gateway buffer implements the notify operation by creating an update event and sending it to the replay buffer.

class CustomsGatewayBuffer...

  public void Notify (DateTime arrivalDate, Ship ship, Port port) {
    CustomsNotificationEvent ev =
        new CustomsNotificationEvent(gateway, arrivalDate, ship, port);
    buffer.Send(ev);
  }
  ICustomsGateway gateway;
  ReplayBuffer buffer;

class CustomsNotificationEvent...

  public CustomsNotificationEvent(
    ICustomsGateway gateway, DateTime arrivalDate, 
    Ship ship, Port port) 
  {
    this.gateway = gateway;
    this.arrivalDate = arrivalDate;
    this.ship = ship;
    this.port = port;
  }

The replay buffer then processes the event if it's active, which leads to a call on the real gateway front.

class ReplayBuffer...

  internal void Send(IGatewayEvent ev) {
    current.Add(ev);
    if (isActive) 
      ev.Process();
  }

class CustomsNotificationEvent...

  public virtual void Process() {
    gateway.Notify(arrivalDate, ship, port) ;
  }

(I'll talk about how a replay buffer becomes active and what the current list is in a moment.)

So as you can see, for active processing of domain events all that happens is the gateway buffer creates a gateway event, sends it to the replay buffer which processes it, which causes the original invocation to be applied to the actual gateway. All of which is an extremely complicated form of simple delegation.

The payoff for this weirdness comes in processing the retroactive events. The replay buffer is connected to the event processor through a rewindable interface.

class ReplayBuffer...

  IRewindable eventProcessor;

internal interface IRewindable  {
  event EventProcessor.EventHandler RewindStarted, RewindFinished, ReplayFinished;    
}

This interface defines three events that matter. The event processor signals these events during its processing of the various cases. Here's the out of order case

class EventProcessor...

  private void ProcessOutOfOrder(DomainEvent e) {
    RewindStarted();
    RewindTo(e);
    RewindFinished();
    BasicProcessEvent(e);
    ReplayAfter(e);
    ReplayFinished();
  }

With replacement the overall start and end are common but each case needs specific placement to indicate when rewind ends.

class EventProcessor...

  private void ProcessReplacement(ReplacementEvent e) {
    RewindStarted();
    if (e.Original.ShouldIgnoreOnReplay)
      throw new ProcessingException("Cannot replace event twice");
    else if (null == e.Replacement) 
      ProcessRejection(e);
    else if (e.HasPriorReplacement) 
      ProcessPriorReplacement(e);
    else 
      ProcessPriorOriginal(e);
    ReplayFinished();
  }

private void ProcessRejection(ReplacementEvent e) {
  RewindTo(e.Original);
  BasicReverseEvent(e.Original);
  RewindFinished();
  e.Original.Reject();
  ReplayAfter(e.Original);
}

private void ProcessPriorOriginal(ReplacementEvent e) {
  RewindTo(e.Original);
  BasicReverseEvent(e.Original);
  RewindFinished();
  e.Original.Reject();
  ReplayBetween(e.Original, e.Replacement);
  BasicProcessEvent(e.Replacement);
  ReplayAfter(e.Replacement);
}

private void ProcessPriorReplacement(ReplacementEvent e) {
  RewindTo(e.Replacement);
  RewindFinished();
  e.Original.Reject();
  BasicProcessEvent(e.Replacement);
  ReplayAfter(e.Replacement);
}

These events give the replay buffer the correct timing information to handle the retroactive processing correctly. I'm using events here because any event processor may have any number of replay buffer doesn't need to know anything about these objects other than that they can understand these three events.

So now lets move to the replay buffer to see how it handles these. The replay buffer is created with a bunch of lists.

class ReplayBuffer...

  internal ReplayBuffer (IRewindable processor, IAdjustable adjuster) {
    this.eventProcessor = processor;
    this.adjuster = adjuster;
    SubscribeToProcessorEvents();
    sent = new ArrayList();
    isActive = true;
    current = sent;
  }
  IList sent,rewound, replayed, current;
  IAdjustable adjuster;
  private bool isActive;
  private void SubscribeToProcessorEvents() {
    eventProcessor.RewindStarted += 
      new EventProcessor.EventHandler(processor_RewindStarted);
    eventProcessor.RewindFinished += 
      new EventProcessor.EventHandler(processor_RewindFinished);
    eventProcessor.ReplayFinished += 
      new EventProcessor.EventHandler(processor_ReplayFinished);
  }

When created the buffer is active, meaning it will process any events sent to it. The current list variable will be set to one of the three core lists, depending on where we are in rewinding - when active it points to the sent list. This gives us a record of all events actually sent on. It also subscribes to all the relevant events on the event processor. (Don't worry about the adjuster for the moment, we'll get there in correction.)

At the beginning of the rewind the buffer makes itself inactive to avoid passing the events on to real gateway, it also swaps the current list variable to point to a fresh list so it can capture all the rewound events.

class ReplayBuffer...

  private void processor_RewindStarted() {
    isActive = false;
    rewound = new ArrayList();
    current = rewound;
  }

With this example you can probably figure out what what happens when the rewind stops, but just in case...

class ReplayBuffer...

  private void processor_RewindFinished() {
    replayed = new ArrayList();
    current = replayed;
  }

So by now it's clear that the last event will be the most interesting as it's here that we need to compute the mismatch lists.

class ReplayBuffer...

  private void processor_ReplayFinished() {
    current = sent;
    DetermineChange();
    isActive = true;
    rewound = null;
    replayed = null;
  }
  private void DetermineChange() {
    IList matchingEvents = new ArrayList();
    foreach (IGatewayEvent ev in replayed) {
      if (rewound.Contains(ev)) matchingEvents.Add(ev);
    }
    foreach (IGatewayEvent ev in matchingEvents) {
      replayed.Remove(ev);
      rewound.Remove(ev);
    }
    adjuster.Adjust(rewound, replayed);
  }

The algorithm is really rather simple, we just find which events match and remove them from both lists. Once we are done we have two lists of mismatched events which we pass on to the adjuster to deal with.

class CustomsGatewayBuffer…

  public void Adjust(IList oldEvents, IList newEvents) {
    foreach (CustomsNotificationEvent e in oldEvents)
      new CustomsCancellationEvent(e).Process();
    foreach (CustomsNotificationEvent e in newEvents)
      e.Process();
  }
class CustomsCancellationEvent {
  private CustomsNotificationEvent original;
  public CustomsCancellationEvent(CustomsNotificationEvent original) {
    this.original = original;
  }
  public void Process() {
    original.Gateway.Cancel(original.ArrivalDate, original.Ship, original.Port);
  }
}

In this case I get the customs gateway buffer to handle the adjustment by simply sending a cancellation notice for all the events that ended up rewound and a new notification for the fresh events. In my fantasy world, government agencies are really very accommodating. In practice each adjustment needs to be considered individually, they could easily get quite complicated and require human intervention. The lists of events, however, should help a great deal to sort out what needs to be done for the external system.