Correlation ID as 1st class citizen

[paneq]

Imagine if dispatcher wrapped calling event handlers in

def call(subscriber, event)
  res.with_metadata(correlation_id: event.metadata[:correlation_id] || event.event_id) do
    subscriber = subscriber.new if Class === subscriber
    subscriber.call(event)
  end
end

That would immediately help in debugging and propagating correlation between initial event and events triggered by it. If the initial event gets correlation_id copied from command then we also know which command/request triggered the whole tree of events.

Question: How to handle async event handlers.

Related:

• "Experiment with metadata-driven streams Experiment with metadata-driven streams #221"
• "with_metadata" Allow to dynamically enrich events metadata by using with_metadata #327
• Storing Commands in RES Storing Commands in RES #340

[paneq]

The are both really simple patterns I have never quite understood why 
they end up so misunderstood. 

Let's say every message has 3 ids. 1 is its id. Another is correlation 
the last it causation. 

The rules are quite simple. If you are responding to a message, you 
copy its correlation id as your correlation id, its message id is your 
causation id. 

This allows you to see an entire conversation (correlation id) or to 
see what causes what (causation id). 

https://groups.google.com/forum/#!topic/event-store/pvk2iEBgBtA

[paneq]

Just as ruby event store automatically fills out the timestimep, it should also automatically fill out causation_id and correlation_id from event.id if they are blank/not provided. That way the 1st event in chain is going to have those 3 ids filled out instead of having ID only.

[paneq]

Also it would nice to have a wrapper for command_bus which would read causation_id and correlation_id and copy them to the command as well. That would make possible to track event->command->event flow as compared to event->event

[paneq]

event.correlate_with(another_event_or_command) could be a nice low-level API to copy those values from another_event_or_command to event's metadata.

• RubyEventStore::Event
• RubyEventStore::Proto

[joelvh]

Should correlation ID and causation ID be individual columns on SQL store? If first class citizens, this would at least make storage and retrieval more efficient. I think of metadata as storage of data for when an individual event is read, whereas I can see querying on the other IDs to get a subset of events.

[paneq]

@joelvh my plan is to link them to streams https://blog.arkency.com/correlation-id-and-causation-id-in-evented-systems/

[joelvh]

Fair enough. I think having some of this behavior as options you can toggle on or off would be ideal, such as automatically linking causation and correlation streams.

[paneq]

@joelvh yep, exactly. I was thinking about an event handler that you can subscribe manually.

[paneq]

Please send feedback for https://github.com/RailsEventStore/rails_event_store/pull/362/files

[joelvh]

I still prefer the individual fields/columns for the correlation ID and causation ID so linking to streams isn’t needed and metadata doesn’t have those fields. However, in keeping to the existing schema and utilizing existing features, I think #362 looks good to me.

[paneq]

I still prefer the individual fields/columns for the correlation ID and causation ID

#362 this is 2 layers above this issues. You can still put them into individual columns if you want to.

[paneq]

• event.correlate_with(event) - documentation
• command ID's API
• event.correlate_with(command)

[paneq]

Related #364 - command.message_id & event.message_id

[paneq]

• Dispatcher wrapped calling event handlers in correlation block

[paneq]

• commands correlated with events which trigger them
• async handlers

[paneq]

PoC:

module RubyEventStore
  class CorrelatedCommands

    def initialize(event_store, command_bus)
      @event_store = event_store
      @command_bus = command_bus
    end

    class MiniEvent < Struct.new(:correlation_id, :message_id)
    end

    def call(command)
      if event_store.metadata[:correlation_id] && event_store.metadata[:causation_id]
        command.correlate_with(MiniEvent.new(
          event_store.metadata[:correlation_id],
          event_store.metadata[:causation_id],
        ))
        event_store.with_metadata(
          correlation_id: event_store.metadata[:correlation_id],
          causation_id: command.message_id,
        ) do
          @command_bus.call(command)
        end
      else
        event_store.with_metadata(
          correlation_id: command.message_id,
          causation_id: command.message_id,
        ) do
          @command_bus.call(command)
        end
      end
    end

  end
end

[paneq]

• async handlers helpers documentation
• Handlers persisting to certain streams based on metadata