<commit_before>module RubyEventStore
  # @private
  class SRecord
    def self.new(
      event_id:   SecureRandom.uuid,
      data:       '{}',
      metadata:   '{}',
      event_type: 'SRecordTestEvent'
    )
      SerializedRecord.new(
        event_id:   event_id,
        data:       data,
        metadata:   metadata,
        event_type: event_type,
      )
    end
  end

  # @private
  Type1 = Class.new(RubyEventStore::Event)
  # @private
  Type2 = Class.new(RubyEventStore::Event)
  # @private
  Type3 = Class.new(RubyEventStore::Event)
end

module RubyEventStore
  RSpec.shared_examples :event_repository do |repository_class, rescuable_concurrency_test_errors = []|
    let(:repository)    { subject || repository_class.new }
    let(:mapper)        { Mappers::NullMapper.new }
    let(:specification) { Specification.new(SpecificationReader.new(repository, mapper)) }
    let(:global_stream) { Stream.new(GLOBAL_STREAM) }
    let(:stream)        { Stream.new(SecureRandom.uuid) }
    let(:stream_flow)   { Stream.new('flow') }
    let(:stream_other)  { Stream.new('other') }
    let(:stream_test)   { Stream.new('test') }
    let(:version_none)  { ExpectedVersion.none }
    let(:version_auto)  { ExpectedVersion.auto }
    let(:version_any)   { ExpectedVersion.any }
    let(:version_0)     { ExpectedVersion.new(0) }
    let(:version_1)     { ExpectedVersion.new(1) }
    let(:version_2)     { ExpectedVersion.new(2) }
    let(:version_3)     { ExpectedVersion.new(3) }

    def read_stream_events_forward(repository, stream)
      repository.read(
        specification
          .stream(stream.name)
          .result
      ).to_a
    end

    def read_stream_events_backward(repository, stream)
      repository.read(
        specification
          .stream(stream.name)
          .backward
          .result
      ).to_a
    end

    def read_events_forward(repository, stream, start, stop, count)
      repository.read(
        specification
          .stream(stream.name)
          .from(start)
          .to(stop)
          .limit(count)
          .result
      ).to_a
    end

    def read_events_backward(repository, stream, start, stop, count)
      repository.read(
        specification
          .stream(stream.name)
          .from(start)
          .to(stop)
          .limit(count)
          .backward
          .result
      ).to_a
    end

    def read_all_streams_forward(repository, start, stop, count)
      repository.read(
        specification
          .from(start)
          .to(stop)
          .limit(count)
          .result
      ).to_a
    end

    def read_all_streams_backward(repository, start, stop, count)
      repository.read(
        specification
          .from(start)
          .to(stop)
          .limit(count)
          .backward
          .result
      ).to_a
    end

    it 'just created is empty' do
      expect(read_all_streams_forward(repository, :begin, :end, 1)).to be_empty
    end

    specify 'append_to_stream returns self' do
      repository.
        append_to_stream(event = SRecord.new, stream, version_none).
        append_to_stream(event = SRecord.new, stream, version_0)
    end

    specify 'link_to_stream returns self' do
      event0 = SRecord.new
      event1 = SRecord.new
      repository.
        append_to_stream([event0, event1], stream, version_none).
        link_to_stream(event0.event_id, stream_flow, version_none).
        link_to_stream(event1.event_id, stream_flow, version_0)
    end

    specify 'adds an initial event to a new stream' do
      repository.append_to_stream(event = SRecord.new, stream, version_none)
      expect(read_all_streams_forward(repository, :begin, :end, 1).first).to eq(event)
      expect(read_stream_events_forward(repository, stream).first).to eq(event)
      expect(read_stream_events_forward(repository, stream_other)).to be_empty
    end

    specify 'links an initial event to a new stream' do
      repository.
        append_to_stream(event = SRecord.new, stream, version_none).
        link_to_stream(event.event_id, stream_flow, version_none)

      expect(read_all_streams_forward(repository, :begin, :end, 1).first).to eq(event)
      expect(read_stream_events_forward(repository, stream).first).to eq(event)
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event])
      expect(read_stream_events_forward(repository, stream_other)).to be_empty
    end

    specify 'adds multiple initial events to a new stream' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      expect(read_all_streams_forward(repository, :begin, :end, 2)).to eq([event0, event1])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1])
    end

    specify 'links multiple initial events to a new stream' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none).link_to_stream([
        event0.event_id,
        event1.event_id,
      ], stream_flow, version_none)
      expect(read_all_streams_forward(repository, :begin, :end, 2)).to eq([event0, event1])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event0, event1])
    end

    specify 'correct expected version on second write' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_1)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1, event2, event3])
    end

    specify 'correct expected version on second link' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none).append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream_flow, version_none).link_to_stream([
        event0.event_id,
        event1.event_id,
      ], stream_flow, version_1)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event2, event3, event0, event1])
    end

    specify 'incorrect expected version on second write' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      expect do
        repository.append_to_stream([
          event2 = SRecord.new,
          event3 = SRecord.new,
        ], stream, version_0)
      end.to raise_error(WrongExpectedEventVersion)

      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1])
    end

    specify 'incorrect expected version on second link' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream_other, version_none)
      expect do
        repository.link_to_stream([
          event2.event_id,
          event3.event_id,
        ], stream, version_0)
      end.to raise_error(WrongExpectedEventVersion)

      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1])
    end

    specify ':none on first and subsequent write' do
      repository.append_to_stream([
        eventA = SRecord.new,
      ], stream, version_none)
      expect do
        repository.append_to_stream([
          eventB = SRecord.new,
        ], stream, version_none)
      end.to raise_error(WrongExpectedEventVersion)
      expect(read_all_streams_forward(repository, :begin, :end, 1)).to eq([eventA])
      expect(read_stream_events_forward(repository, stream)).to eq([eventA])
    end

    specify ':none on first and subsequent link' do
      repository.append_to_stream([
        eventA = SRecord.new,
        eventB = SRecord.new,
      ], stream, version_none)

      repository.link_to_stream([eventA.event_id], stream_flow, version_none)
      expect do
        repository.link_to_stream([eventB.event_id], stream_flow, version_none)
      end.to raise_error(WrongExpectedEventVersion)

      expect(read_all_streams_forward(repository, :begin, :end, 1)).to eq([eventA])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([eventA])
    end

    specify ':any allows stream with best-effort order and no guarantee' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_any)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_any)
      expect(read_all_streams_forward(repository, :begin, :end, 4).to_set).to eq(Set.new([event0, event1, event2, event3]))
      expect(read_stream_events_forward(repository, stream).to_set).to eq(Set.new([event0, event1, event2, event3]))
    end

    specify ':any allows linking in stream with best-effort order and no guarantee' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_any)

      repository.link_to_stream([
        event0.event_id, event1.event_id,
      ], stream_flow, version_any)
      repository.link_to_stream([
        event2.event_id, event3.event_id,
      ], stream_flow, version_any)

      expect(read_all_streams_forward(repository, :begin, :end, 4).to_set).to eq(Set.new([event0, event1, event2, event3]))
      expect(read_stream_events_forward(repository, stream_flow).to_set).to eq(Set.new([event0, event1, event2, event3]))
    end

    specify ':auto queries for last position in given stream' do
      repository.append_to_stream([
        eventA = SRecord.new,
        eventB = SRecord.new,
        eventC = SRecord.new,
      ], stream_other, version_auto)
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_1)
    end

    specify ':auto queries for last position in given stream when linking' do
      repository.append_to_stream([
        eventA = SRecord.new,
        eventB = SRecord.new,
        eventC = SRecord.new,
      ], stream_other, version_auto)
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.link_to_stream([
        eventA.event_id,
        eventB.event_id,
        eventC.event_id,
      ], stream, version_1)
    end

    specify ':auto starts from 0' do
      repository.append_to_stream([
        event0 = SRecord.new,
      ], stream, version_auto)
      expect do
        repository.append_to_stream([
          event1 = SRecord.new,
        ], stream, version_none)
      end.to raise_error(WrongExpectedEventVersion)
    end

    specify ':auto linking starts from 0' do
      repository.append_to_stream([
        event0 = SRecord.new,
      ], stream_other, version_auto)
      repository.link_to_stream([
        event0.event_id,
      ], stream, version_auto)
      expect do
        repository.append_to_stream([
          event1 = SRecord.new,
        ], stream, version_none)
      end.to raise_error(WrongExpectedEventVersion)
    end

    specify ':auto queries for last position and follows in incremental way' do
      # It is expected that there is higher level lock
      # So this query is safe from race conditions
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_auto)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([
        event0, event1,
        event2, event3
      ])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1, event2, event3])
    end

    specify ':auto queries for last position and follows in incremental way when linking' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_auto)
      repository.link_to_stream([
        event0.event_id, event1.event_id,
      ], stream_flow, version_auto)
      repository.link_to_stream([
        event2.event_id, event3.event_id,
      ], stream_flow, version_auto)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([
        event0, event1,
        event2, event3
      ])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event0, event1, event2, event3])
    end

    specify ':auto is compatible with manual expectation' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_1)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1, event2, event3])
    end

    specify ':auto is compatible with manual expectation when linking' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.link_to_stream([
        event0.event_id,
      ], stream_flow, version_auto)
      repository.link_to_stream([
        event1.event_id,
      ], stream_flow, version_0)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1,])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event0, event1,])
    end

    specify 'manual is compatible with auto expectation' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_auto)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1, event2, event3])
    end

    specify 'manual is compatible with auto expectation when linking' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.link_to_stream([
        event0.event_id,
      ], stream_flow, version_none)
      repository.link_to_stream([
        event1.event_id,
      ], stream_flow, version_auto)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event0, event1])
    end

    specify 'unlimited concurrency for :any - everything should succeed', timeout: 10, mutant: false do
      skip unless test_race_conditions_any
      verify_conncurency_assumptions
      begin
        concurrency_level = 4
        fail_occurred     = false
        wait_for_it       = true

        threads = concurrency_level.times.map do |i|
          Thread.new do
            true while wait_for_it
            begin
              100.times do |j|
                eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
                repository.append_to_stream([
                  SRecord.new(event_id: eid),
                ], stream, version_any)
              end
            rescue WrongExpectedEventVersion
              fail_occurred = true
            end
          end
        end
        wait_for_it = false
        threads.each(&:join)
        expect(fail_occurred).to eq(false)
        expect(read_stream_events_forward(repository, stream).size).to eq(400)
        events_in_stream = read_stream_events_forward(repository, stream)
        expect(events_in_stream.size).to eq(400)
        events0 = events_in_stream.select do |ev|
          ev.event_id.start_with?("0-")
        end
        expect(events0).to eq(events0.sort_by{|ev| ev.event_id })
      ensure
        cleanup_concurrency_test
      end
    end

    specify 'unlimited concurrency for :any - everything should succeed when linking', timeout: 10, mutant: false do
      skip unless test_race_conditions_any
      verify_conncurency_assumptions
      begin
        concurrency_level = 4
        fail_occurred     = false
        wait_for_it       = true

        concurrency_level.times.map do |i|
          100.times do |j|
            eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
            repository.append_to_stream([
              SRecord.new(event_id: eid),
            ], stream, version_any)
          end
        end

        threads = concurrency_level.times.map do |i|
          Thread.new do
            true while wait_for_it
            begin
              100.times do |j|
                eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
                repository.link_to_stream(eid, stream_flow, version_any)
              end
            rescue WrongExpectedEventVersion
              fail_occurred = true
            end
          end
        end
        wait_for_it = false
        threads.each(&:join)
        expect(fail_occurred).to eq(false)
        expect(read_stream_events_forward(repository, stream_flow).size).to eq(400)
        events_in_stream = read_stream_events_forward(repository, stream_flow)
        expect(events_in_stream.size).to eq(400)
        events0 = events_in_stream.select do |ev|
          ev.event_id.start_with?("0-")
        end
        expect(events0).to eq(events0.sort_by{|ev| ev.event_id })
      ensure
        cleanup_concurrency_test
      end
    end

    specify 'limited concurrency for :auto - some operations will fail without outside lock, stream is ordered', mutant: false do
      skip unless test_race_conditions_auto
      verify_conncurency_assumptions
      begin
        concurrency_level = 4

        fail_occurred = 0
        wait_for_it  = true

        threads = concurrency_level.times.map do |i|
          Thread.new do
            Thread.current.report_on_exception = false
            true while wait_for_it
            100.times do |j|
              begin
                eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
                repository.append_to_stream([
                  SRecord.new(event_id: eid),
                ], stream, version_auto)
                sleep(rand(concurrency_level) / 1000.0)
              rescue WrongExpectedEventVersion, *rescuable_concurrency_test_errors
                fail_occurred +=1
              end
            end
          end
        end
        wait_for_it = false
        threads.each(&:join)
        expect(fail_occurred).to be > 0
        events_in_stream = read_stream_events_forward(repository, stream)
        expect(events_in_stream.size).to be < 400
        expect(events_in_stream.size).to be >= 100
        events0 = events_in_stream.select do |ev|
          ev.event_id.start_with?("0-")
        end
        expect(events0).to eq(events0.sort_by{|ev| ev.event_id })
        additional_limited_concurrency_for_auto_check
      ensure
        cleanup_concurrency_test
      end
    end

    specify 'limited concurrency for :auto - some operations will fail without outside lock, stream is ordered', mutant: false do
      skip unless test_race_conditions_auto
      verify_conncurency_assumptions
      begin
        concurrency_level = 4

        concurrency_level.times.map do |i|
          100.times do |j|
            eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
            repository.append_to_stream([
              SRecord.new(event_id: eid),
            ], stream_other, version_any)
          end
        end

        fail_occurred = 0
        wait_for_it  = true

        threads = concurrency_level.times.map do |i|
          Thread.new do
            Thread.current.report_on_exception = false
            true while wait_for_it<commit_msg>Superfluous silencers.<commit_after>module RubyEventStore
  # @private
  class SRecord
    def self.new(
      event_id:   SecureRandom.uuid,
      data:       '{}',
      metadata:   '{}',
      event_type: 'SRecordTestEvent'
    )
      SerializedRecord.new(
        event_id:   event_id,
        data:       data,
        metadata:   metadata,
        event_type: event_type,
      )
    end
  end

  # @private
  Type1 = Class.new(RubyEventStore::Event)
  # @private
  Type2 = Class.new(RubyEventStore::Event)
  # @private
  Type3 = Class.new(RubyEventStore::Event)
end

module RubyEventStore
  RSpec.shared_examples :event_repository do |repository_class, rescuable_concurrency_test_errors = []|
    let(:repository)    { subject || repository_class.new }
    let(:mapper)        { Mappers::NullMapper.new }
    let(:specification) { Specification.new(SpecificationReader.new(repository, mapper)) }
    let(:global_stream) { Stream.new(GLOBAL_STREAM) }
    let(:stream)        { Stream.new(SecureRandom.uuid) }
    let(:stream_flow)   { Stream.new('flow') }
    let(:stream_other)  { Stream.new('other') }
    let(:stream_test)   { Stream.new('test') }
    let(:version_none)  { ExpectedVersion.none }
    let(:version_auto)  { ExpectedVersion.auto }
    let(:version_any)   { ExpectedVersion.any }
    let(:version_0)     { ExpectedVersion.new(0) }
    let(:version_1)     { ExpectedVersion.new(1) }
    let(:version_2)     { ExpectedVersion.new(2) }
    let(:version_3)     { ExpectedVersion.new(3) }

    def read_stream_events_forward(repository, stream)
      repository.read(
        specification
          .stream(stream.name)
          .result
      ).to_a
    end

    def read_stream_events_backward(repository, stream)
      repository.read(
        specification
          .stream(stream.name)
          .backward
          .result
      ).to_a
    end

    def read_events_forward(repository, stream, start, stop, count)
      repository.read(
        specification
          .stream(stream.name)
          .from(start)
          .to(stop)
          .limit(count)
          .result
      ).to_a
    end

    def read_events_backward(repository, stream, start, stop, count)
      repository.read(
        specification
          .stream(stream.name)
          .from(start)
          .to(stop)
          .limit(count)
          .backward
          .result
      ).to_a
    end

    def read_all_streams_forward(repository, start, stop, count)
      repository.read(
        specification
          .from(start)
          .to(stop)
          .limit(count)
          .result
      ).to_a
    end

    def read_all_streams_backward(repository, start, stop, count)
      repository.read(
        specification
          .from(start)
          .to(stop)
          .limit(count)
          .backward
          .result
      ).to_a
    end

    it 'just created is empty' do
      expect(read_all_streams_forward(repository, :begin, :end, 1)).to be_empty
    end

    specify 'append_to_stream returns self' do
      repository.
        append_to_stream(event = SRecord.new, stream, version_none).
        append_to_stream(event = SRecord.new, stream, version_0)
    end

    specify 'link_to_stream returns self' do
      event0 = SRecord.new
      event1 = SRecord.new
      repository.
        append_to_stream([event0, event1], stream, version_none).
        link_to_stream(event0.event_id, stream_flow, version_none).
        link_to_stream(event1.event_id, stream_flow, version_0)
    end

    specify 'adds an initial event to a new stream' do
      repository.append_to_stream(event = SRecord.new, stream, version_none)
      expect(read_all_streams_forward(repository, :begin, :end, 1).first).to eq(event)
      expect(read_stream_events_forward(repository, stream).first).to eq(event)
      expect(read_stream_events_forward(repository, stream_other)).to be_empty
    end

    specify 'links an initial event to a new stream' do
      repository.
        append_to_stream(event = SRecord.new, stream, version_none).
        link_to_stream(event.event_id, stream_flow, version_none)

      expect(read_all_streams_forward(repository, :begin, :end, 1).first).to eq(event)
      expect(read_stream_events_forward(repository, stream).first).to eq(event)
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event])
      expect(read_stream_events_forward(repository, stream_other)).to be_empty
    end

    specify 'adds multiple initial events to a new stream' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      expect(read_all_streams_forward(repository, :begin, :end, 2)).to eq([event0, event1])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1])
    end

    specify 'links multiple initial events to a new stream' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none).link_to_stream([
        event0.event_id,
        event1.event_id,
      ], stream_flow, version_none)
      expect(read_all_streams_forward(repository, :begin, :end, 2)).to eq([event0, event1])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event0, event1])
    end

    specify 'correct expected version on second write' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_1)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1, event2, event3])
    end

    specify 'correct expected version on second link' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none).append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream_flow, version_none).link_to_stream([
        event0.event_id,
        event1.event_id,
      ], stream_flow, version_1)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event2, event3, event0, event1])
    end

    specify 'incorrect expected version on second write' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      expect do
        repository.append_to_stream([
          event2 = SRecord.new,
          event3 = SRecord.new,
        ], stream, version_0)
      end.to raise_error(WrongExpectedEventVersion)

      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1])
    end

    specify 'incorrect expected version on second link' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream_other, version_none)
      expect do
        repository.link_to_stream([
          event2.event_id,
          event3.event_id,
        ], stream, version_0)
      end.to raise_error(WrongExpectedEventVersion)

      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1])
    end

    specify ':none on first and subsequent write' do
      repository.append_to_stream([
        eventA = SRecord.new,
      ], stream, version_none)
      expect do
        repository.append_to_stream([
          eventB = SRecord.new,
        ], stream, version_none)
      end.to raise_error(WrongExpectedEventVersion)
      expect(read_all_streams_forward(repository, :begin, :end, 1)).to eq([eventA])
      expect(read_stream_events_forward(repository, stream)).to eq([eventA])
    end

    specify ':none on first and subsequent link' do
      repository.append_to_stream([
        eventA = SRecord.new,
        eventB = SRecord.new,
      ], stream, version_none)

      repository.link_to_stream([eventA.event_id], stream_flow, version_none)
      expect do
        repository.link_to_stream([eventB.event_id], stream_flow, version_none)
      end.to raise_error(WrongExpectedEventVersion)

      expect(read_all_streams_forward(repository, :begin, :end, 1)).to eq([eventA])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([eventA])
    end

    specify ':any allows stream with best-effort order and no guarantee' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_any)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_any)
      expect(read_all_streams_forward(repository, :begin, :end, 4).to_set).to eq(Set.new([event0, event1, event2, event3]))
      expect(read_stream_events_forward(repository, stream).to_set).to eq(Set.new([event0, event1, event2, event3]))
    end

    specify ':any allows linking in stream with best-effort order and no guarantee' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_any)

      repository.link_to_stream([
        event0.event_id, event1.event_id,
      ], stream_flow, version_any)
      repository.link_to_stream([
        event2.event_id, event3.event_id,
      ], stream_flow, version_any)

      expect(read_all_streams_forward(repository, :begin, :end, 4).to_set).to eq(Set.new([event0, event1, event2, event3]))
      expect(read_stream_events_forward(repository, stream_flow).to_set).to eq(Set.new([event0, event1, event2, event3]))
    end

    specify ':auto queries for last position in given stream' do
      repository.append_to_stream([
        eventA = SRecord.new,
        eventB = SRecord.new,
        eventC = SRecord.new,
      ], stream_other, version_auto)
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_1)
    end

    specify ':auto queries for last position in given stream when linking' do
      repository.append_to_stream([
        eventA = SRecord.new,
        eventB = SRecord.new,
        eventC = SRecord.new,
      ], stream_other, version_auto)
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.link_to_stream([
        eventA.event_id,
        eventB.event_id,
        eventC.event_id,
      ], stream, version_1)
    end

    specify ':auto starts from 0' do
      repository.append_to_stream([
        event0 = SRecord.new,
      ], stream, version_auto)
      expect do
        repository.append_to_stream([
          event1 = SRecord.new,
        ], stream, version_none)
      end.to raise_error(WrongExpectedEventVersion)
    end

    specify ':auto linking starts from 0' do
      repository.append_to_stream([
        event0 = SRecord.new,
      ], stream_other, version_auto)
      repository.link_to_stream([
        event0.event_id,
      ], stream, version_auto)
      expect do
        repository.append_to_stream([
          event1 = SRecord.new,
        ], stream, version_none)
      end.to raise_error(WrongExpectedEventVersion)
    end

    specify ':auto queries for last position and follows in incremental way' do
      # It is expected that there is higher level lock
      # So this query is safe from race conditions
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_auto)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([
        event0, event1,
        event2, event3
      ])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1, event2, event3])
    end

    specify ':auto queries for last position and follows in incremental way when linking' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_auto)
      repository.link_to_stream([
        event0.event_id, event1.event_id,
      ], stream_flow, version_auto)
      repository.link_to_stream([
        event2.event_id, event3.event_id,
      ], stream_flow, version_auto)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([
        event0, event1,
        event2, event3
      ])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event0, event1, event2, event3])
    end

    specify ':auto is compatible with manual expectation' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_1)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1, event2, event3])
    end

    specify ':auto is compatible with manual expectation when linking' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.link_to_stream([
        event0.event_id,
      ], stream_flow, version_auto)
      repository.link_to_stream([
        event1.event_id,
      ], stream_flow, version_0)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1,])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event0, event1,])
    end

    specify 'manual is compatible with auto expectation' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_none)
      repository.append_to_stream([
        event2 = SRecord.new,
        event3 = SRecord.new,
      ], stream, version_auto)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1, event2, event3])
      expect(read_stream_events_forward(repository, stream)).to eq([event0, event1, event2, event3])
    end

    specify 'manual is compatible with auto expectation when linking' do
      repository.append_to_stream([
        event0 = SRecord.new,
        event1 = SRecord.new,
      ], stream, version_auto)
      repository.link_to_stream([
        event0.event_id,
      ], stream_flow, version_none)
      repository.link_to_stream([
        event1.event_id,
      ], stream_flow, version_auto)
      expect(read_all_streams_forward(repository, :begin, :end, 4)).to eq([event0, event1])
      expect(read_stream_events_forward(repository, stream_flow)).to eq([event0, event1])
    end

    specify 'unlimited concurrency for :any - everything should succeed', timeout: 10, mutant: false do
      skip unless test_race_conditions_any
      verify_conncurency_assumptions
      begin
        concurrency_level = 4
        fail_occurred     = false
        wait_for_it       = true

        threads = concurrency_level.times.map do |i|
          Thread.new do
            true while wait_for_it
            begin
              100.times do |j|
                eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
                repository.append_to_stream([
                  SRecord.new(event_id: eid),
                ], stream, version_any)
              end
            rescue WrongExpectedEventVersion
              fail_occurred = true
            end
          end
        end
        wait_for_it = false
        threads.each(&:join)
        expect(fail_occurred).to eq(false)
        expect(read_stream_events_forward(repository, stream).size).to eq(400)
        events_in_stream = read_stream_events_forward(repository, stream)
        expect(events_in_stream.size).to eq(400)
        events0 = events_in_stream.select do |ev|
          ev.event_id.start_with?("0-")
        end
        expect(events0).to eq(events0.sort_by{|ev| ev.event_id })
      ensure
        cleanup_concurrency_test
      end
    end

    specify 'unlimited concurrency for :any - everything should succeed when linking', timeout: 10, mutant: false do
      skip unless test_race_conditions_any
      verify_conncurency_assumptions
      begin
        concurrency_level = 4
        fail_occurred     = false
        wait_for_it       = true

        concurrency_level.times.map do |i|
          100.times do |j|
            eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
            repository.append_to_stream([
              SRecord.new(event_id: eid),
            ], stream, version_any)
          end
        end

        threads = concurrency_level.times.map do |i|
          Thread.new do
            true while wait_for_it
            begin
              100.times do |j|
                eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
                repository.link_to_stream(eid, stream_flow, version_any)
              end
            rescue WrongExpectedEventVersion
              fail_occurred = true
            end
          end
        end
        wait_for_it = false
        threads.each(&:join)
        expect(fail_occurred).to eq(false)
        expect(read_stream_events_forward(repository, stream_flow).size).to eq(400)
        events_in_stream = read_stream_events_forward(repository, stream_flow)
        expect(events_in_stream.size).to eq(400)
        events0 = events_in_stream.select do |ev|
          ev.event_id.start_with?("0-")
        end
        expect(events0).to eq(events0.sort_by{|ev| ev.event_id })
      ensure
        cleanup_concurrency_test
      end
    end

    specify 'limited concurrency for :auto - some operations will fail without outside lock, stream is ordered', mutant: false do
      skip unless test_race_conditions_auto
      verify_conncurency_assumptions
      begin
        concurrency_level = 4

        fail_occurred = 0
        wait_for_it  = true

        threads = concurrency_level.times.map do |i|
          Thread.new do
            true while wait_for_it
            100.times do |j|
              begin
                eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
                repository.append_to_stream([
                  SRecord.new(event_id: eid),
                ], stream, version_auto)
                sleep(rand(concurrency_level) / 1000.0)
              rescue WrongExpectedEventVersion, *rescuable_concurrency_test_errors
                fail_occurred +=1
              end
            end
          end
        end
        wait_for_it = false
        threads.each(&:join)
        expect(fail_occurred).to be > 0
        events_in_stream = read_stream_events_forward(repository, stream)
        expect(events_in_stream.size).to be < 400
        expect(events_in_stream.size).to be >= 100
        events0 = events_in_stream.select do |ev|
          ev.event_id.start_with?("0-")
        end
        expect(events0).to eq(events0.sort_by{|ev| ev.event_id })
        additional_limited_concurrency_for_auto_check
      ensure
        cleanup_concurrency_test
      end
    end

    specify 'limited concurrency for :auto - some operations will fail without outside lock, stream is ordered', mutant: false do
      skip unless test_race_conditions_auto
      verify_conncurency_assumptions
      begin
        concurrency_level = 4

        concurrency_level.times.map do |i|
          100.times do |j|
            eid = "0000000#{i}-#{sprintf("%04d", j)}-0000-0000-000000000000"
            repository.append_to_stream([
              SRecord.new(event_id: eid),
            ], stream_other, version_any)
          end
        end

        fail_occurred = 0
        wait_for_it  = true

        threads = concurrency_level.times.map do |i|
          Thread.new do
            true while wait_for_it
{'fraction_non_alphanumeric': 0.09599555866663448, 'fraction_numerical': 0.020372203046175385, 'mean_word_length': 2.534681341182493, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 20, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5527847', 'n_tokens_mistral': 12964, 'n_tokens_neox': 12473, 'n_words': 2642}