Event Store as queue

[paneq]

Implement Event Store in a way that would allow clients to safely iterate over global stream of all events with pagination and no race conditions. The client would remember its position in the stream of events in its own side.

[paneq]

By safely I mean in a way which allows to avoid this potential problem when reading events:

In such case we can get IDs of events 0 and 2, but skip reading uncommitted event nr 1.

We would like to avoid such problems.

From my reading about it and thinking over it, I see 2 potential solutions:

• linearize writing to EventStore by using exclusive lock on underlying DB. This can however limit write throughput which I would like to avoid.
• The other solution I thought about is different. It would delay the problem until read.
  • The problem is that assuming auto-incremented IDs of events in global stream there can be a gap in read data. We don't easily know if that data was not yet committed or rolled back.
  • what if we used after_commit to marked rolled-backed positions as taken so the reading process would know to skip them
  • but after_commit can fail when server is turned off. What if we had a constantly running external process that would try to fill the gaps with missing numbers. If the transaction was rolled-back it would simply fill the missing number. If the transaction was long and still running it would cause a deadlock (ideally in a way that it would loose and keep trying until the competing transaction is committed or rolled-back)
  • that supervisor process theoretically should be needed very rarely. Only when there was rollback and server was turned off so the usual after_commit callbacks did not have enough time to finish.

In case of a very long transaction with events, the write throughput would not be affected but a separate queue reading process would not be able to proceed until the transaction is finished. As very long transactions don't happen often it might not be a problem every in an app.

[paneq]

Imagine starting point like this:

require 'active_record'

ENV['DATABASE_URL'] ||= "postgres://localhost/rails_event_store_active_record?pool=5"

ActiveRecord::Base.establish_connection(ENV['DATABASE_URL'])
ActiveRecord::Schema.define do
  self.verbose = true

  enable_extension "plpgsql"
  enable_extension "pgcrypto"

  create_table(:event_store_events, id: :uuid, force: true) do |t|
    t.string      :event_type,  null: false
    t.text        :metadata
    t.text        :data,        null: false
    t.datetime    :created_at,  null: false
  end
  add_index :event_store_events, :created_at

  create_table(:event_store_global_stream, force: true) do |t|
    t.references :event, null: true, type: :uuid
  end

  create_table(:event_store_events_in_streams, force: true) do |t|
    t.string      :stream,      null: false
    t.integer     :position,    null: true
    t.references  :event, null: false, type: :uuid
    t.datetime    :created_at,  null: false
  end
  add_index :event_store_events_in_streams, [:stream, :position], unique: true
  add_index :event_store_events_in_streams, [:created_at]
end

ActiveRecord::Base.logger = Logger.new(STDOUT)

class Event < ::ActiveRecord::Base
  self.primary_key = :id
  self.table_name = 'event_store_events'
  serialize :metadata
  serialize :data
end

class EventInStream < ::ActiveRecord::Base
  self.primary_key = :id
  self.table_name = 'event_store_events_in_streams'
  belongs_to :event
end

class EventInGlobalStream < ::ActiveRecord::Base
  self.primary_key = :id
  self.table_name = 'event_store_global_stream'
  belongs_to :event

  after_rollback do
    begin
      puts id
      EventInGlobalStreamFillIn.new(id: id).save!
    rescue => e
      puts e.message
      puts e.backtrace.join("\n")
    end
  end
end

class EventInGlobalStreamFillIn < ::ActiveRecord::Base
  self.primary_key = :id
  self.table_name = 'event_store_global_stream'
end

event nr 1 in global stream was inserted:

egs = nil
ActiveRecord::Base.transaction do
  e = Event.new
  e.event_type = "event_type"
  e.data = {}
  e.metadata = {}
  e.save!

  egs = EventInGlobalStream.new(event_id: e.id)
  egs.save!
end

event nr 2 is trying to be inserted but it takes a lot of time:

ActiveRecord::Base.transaction do
  e = Event.new
  e.event_type = "event_type"
  e.data = {}
  e.metadata = {}
  e.save!

  egs = EventInGlobalStream.new(event_id: e.id)
  egs.save!
  sleep(5000)
end

event nr 3 was inserted:

egs = nil
ActiveRecord::Base.transaction do
  e = Event.new
  e.event_type = "event_type"
  e.data = {}
  e.metadata = {}
  e.save!

  egs = EventInGlobalStream.new(event_id: e.id)
  egs.save!
end

The supervisor process responsible for filling gaps would try to execute

ActiveRecord::Base.transaction do
  ActiveRecord::Base.connection.execute "SET LOCAL lock_timeout = '1s';"
  EventInGlobalStreamFillIn.create!(id: 2)
end

It will continue failing until the long transaction is commited or rolled-back.
When it is committed the supervisor will find the gap is no longer there and do nothing, if it is rolled-back without any crash nr 2 will be added without a supervisor. If it is rolledbacked and there is a crash there won't be any deadlock anymore and the supervisor will fill in the missing gap for position nr 2.

The process/logic responsible for exposing the global event log won't return anything >= 2 when there is a gap.

I think that way:

• there is no throughput limitation on write from the fact that we want to suppor using event store as queue.
• we can safely fill in gaps in monotonically growing positions of global stream in case of rolledback transactions
  • even in case of crash
• the worst effect is that the synchronization by reading the global stream will be delayed in case of concurrent transactions when they overlap in taken numbers.

@mlomnicki @mpraglowski what do you think

[paneq]

The solution presented above show it should be possible in Postgres, haven't check how MySQL would behave in such situation

[paneq]

The global stream would be quite disconnected from normal streams. Because of its importance it would have its own dedicated table to be able to run the IDs trick.

Hmm but maybe that's not even necessary. It was easier to consider it being implemented that way to me and write such a proof of concept.

[mlomnicki]

@paneq I need to read it more carefully but at the first sight this solution seems best to me

linearize writing to EventStore by using exclusive lock on underlying DB. This can however limit write throughput which I would like to avoid.

I'd go with this solution. Whether it would be an optimistic or a pessimistic lock doesn't matter but if we want to avoid race conditions we have to lock the stream. Obviously an optimistic lock would be preferred.

Anyway your solution looks interesting and I'd be happy to give it a try and a closer look. Maybe it would be best to implement it as an alternative rails_event_store_active_record? By default RES would offer a simple persistence model but everyone would be free to switch to a faster solution simply by using another repository.

I imagine that ideally we'd like to be both free of race conditions and super fast but in my humble opinion we must pick only one. It'll be either consistent or fast*, not both.

As I already mentioned in another thread, if we want fast writes then maybe we can introduce a special mode or a special type of a stream which wouldn't give any guarantees on the consistency but would offer better throughput. We could offer 2 strategies

• consistent but slow (default)
• fast but inconsistent

then it'd be up to the developer to pick whatever works best to them in the given context

*[obviously"fast" is very relative and subjective...]

[paneq]

Situation:

require 'active_record'
ENV['DATABASE_URL'] ||= "postgres://localhost/rails_event_store_active_record?pool=5"
ActiveRecord::Base.establish_connection(ENV['DATABASE_URL'])

class Event < ActiveRecord::Base
  self.table_name = "event_store_events"
end

# console 1
e = Event.new(event_type: "1", data: "")
e.id = "1"
e.save!

#console 2 - in progress - uncommited
ActiveRecord::Base.transaction do
  e = Event.new(event_type: "1", data: "")
  e.id = "3"
  e.save!
  sleep(50_000_000)
end

#console 3 - done after console 1
e = Event.new(event_type: "1", data: "")
e.id = "4"
e.save!

Let's say we iterate over events ascending starting from 0, we see a gap in position 2. We need to know if that transaction is in progress or rolled-back. If in progress we should only send numbers up to 1 (including). If rolled-backed, we can send 1,3.

Hypothesis

r = ActiveRecord::Base.connection.execute "SELECT id, xmin, xmax from event_store_events"
r.each{|x| puts x}
# {"id"=>"1", "xmin"=>"22085", "xmax"=>"0"}
# {"id"=>"3", "xmin"=>"22088", "xmax"=>"0"}

xminc = "SELECT * FROM txid_snapshot_xmin(txid_current_snapshot());"
r = ActiveRecord::Base.connection.execute(xminc)
r.each{|x| puts x}
# => {"txid_snapshot_xmin"=>22086}

We can only read up to xmin < 22086 as this transaction have not yet finished. It's xmin of current snapshot.

Now I interrupt the ongoing transaction from console 2.

xminc = "SELECT * FROM txid_snapshot_xmin(txid_current_snapshot());"
r = ActiveRecord::Base.connection.execute(xminc)
r.each{|x| puts x}
# {"txid_snapshot_xmin"=>22089}

r = ActiveRecord::Base.connection.execute "SELECT id, xmin, xmax from event_store_events"
r.each{|x| puts x}
# {"id"=>"1", "xmin"=>"22085", "xmax"=>"0"}
# {"id"=>"3", "xmin"=>"22088", "xmax"=>"0"}

Now we know we there will be gap at id=2 because the transaction was committed. and txid_snapshot_xmin(txid_current_snapshot()) is 22089

Based on:

• https://stackoverflow.com/questions/20195088/how-to-spy-on-a-transaction-that-is-not-yet-committed-in-postgres-for-debugging
• https://www.postgresql.org/docs/current/static/ddl-system-columns.html
• https://www.postgresql.org/docs/current/static/functions-info.html#FUNCTIONS-TXID-SNAPSHOT

Perhaps I am mistaken but it seems to me we could just do:

SELECT * 
FROM event_store_events_in_streams
ORDER BY id ASC
WHERE stream= RubyEventStore::GLOBAL_STREAM
AND
id > last_seen_id
AND
xmin < txid_snapshot_xmin(txid_current_snapshot())

if that's possible.

If xmin < txid_snapshot_xmin(txid_current_snapshot()) is not possible maybe we could filter out that part in Ruby up to first record which fails.

P.S. I tested on event_store_events because it was easier to verify this hypothesis but in reality we care about global stream in event_store_events_in_streams.

Dragons

• This number resets after many many transactions ;) We need to handle it becoming lower somehow.

[paneq]

"SELECT * FROM txid_snapshot_xmin(txid_current_snapshot());" and "SELECT id, xmin, xmax from event_store_events" verified to work on Heroku

[paneq]

xmin < in WHERE won't work because:

As an aside, be wary that their type is xid. That's a very capricious version of integer that only responds to = as an operator. Also, it can only be cast to text and back:

so this part would need to be implemented in Ruby. Fortuntely that's easy.

[swistak35]

Sounds good to me.

One detail: this is all under the assumption that we won't update events rows, right? I am asking because following statement in xmin description of postgresql:

each update of a row creates a new row version

So I imagine, that if I would start some longer process which is updating old event (let's say, 1 year old event), then xmin on that row would be very high (meaning: row is "very new"), which could possibly block processing all newer events. But I guess that this is a matter of higher-level implementation, because I imagine that in case, for example, read models, we store somewhere (in read model row?) the event id which was processed lastly and don't even search for such old events.

[paneq]

Other challenges found. Research is continued.

[paneq]

TLDR in Polish of all my research: https://youtu.be/xJpEOCiyJxw

[paneq]

As far as I understand the only ways to get the list of records in the order of committed (without doing any delete or update operations per every synchronizing client) is:

• using Write-Ahead-Log from the database
  • use http://debezium.io/docs/tutorial/ to stream it to Kafka
  • requires loading some plugins
  • requires much DevOps
  • not allowed on some environments
• using track_commit_timestamp in postgresql
  • not allowed on some environments
  • could get wrong with time issues on the server
  • SELECT pg_xact_commit_timestamp(xmin), * FROM events WHERE pg_xact_commit_timestamp(xmin) > SELECT pg_xact_commit_timestamp(SELECT xmin from events WHERE ID = last_seen_id))
  • probably no easy way to use indexes to get only newest records

You can linearize your writes to achieve the properties that we needed - #403