From 88f61eb8c6e557041baecc0099f983a6a3f2f3b8 Mon Sep 17 00:00:00 2001 From: Benjamin Moss Date: Thu, 19 Sep 2024 11:15:06 -0400 Subject: [PATCH 1/3] Include slow tests as they are not that much slower And we've missed seeing some failing tests by not running them. See issue #283 --- test/test_helper.exs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/test_helper.exs b/test/test_helper.exs index 162cd35b..0d7c56fe 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -1,3 +1,3 @@ -exclude = [:ignore, :manual, :migration, :slow] +exclude = [:ignore, :manual, :migration] ExUnit.start(exclude: exclude) From bbd969008e5ec70408267f300d5a4bb7648b88ff Mon Sep 17 00:00:00 2001 From: Benjamin Moss Date: Thu, 19 Sep 2024 11:13:49 -0400 Subject: [PATCH 2/3] Surface original error as we can get to this branch after all --- lib/event_store/streams/stream.ex | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/lib/event_store/streams/stream.ex b/lib/event_store/streams/stream.ex index f82d89b5..e5e87cef 100644 --- a/lib/event_store/streams/stream.ex +++ b/lib/event_store/streams/stream.ex @@ -344,9 +344,7 @@ defmodule EventStore.Streams.Stream do append_to_stream(conn, stream_uuid, expected_version, events, opts) else - # We should never get here, but just in case we break something in another - # part of the app, this will give us better output in the tests. - {:error, :already_retried_once} + {:error, {:already_retried_once, :duplicate_stream_uuid}} end end From 0956c4c92c0e1598faf5fac6b703f577022bfcfb Mon Sep 17 00:00:00 2001 From: Benjamin Moss Date: Thu, 19 Sep 2024 11:12:28 -0400 Subject: [PATCH 3/3] Addresses #283 - Remember the stream_id from batch to batch So that subsequent batches update instead of trying to insert and fail --- .../sql/statements/insert_events.sql.eex | 36 ++++++++++--------- .../insert_events_any_version.sql.eex | 36 ++++++++++--------- lib/event_store/storage/appender.ex | 21 +++++++---- 3 files changed, 54 insertions(+), 39 deletions(-) diff --git a/lib/event_store/sql/statements/insert_events.sql.eex b/lib/event_store/sql/statements/insert_events.sql.eex index 49bcae15..3d444fca 100644 --- a/lib/event_store/sql/statements/insert_events.sql.eex +++ b/lib/event_store/sql/statements/insert_events.sql.eex @@ -57,20 +57,24 @@ WITH SET stream_version = stream_version + $2::bigint WHERE stream_id = 0 RETURNING stream_version - $2::bigint as initial_stream_version + ), + linked_stream_events AS ( + INSERT INTO "<%= schema %>".stream_events + ( + event_id, + stream_id, + stream_version, + original_stream_id, + original_stream_version + ) + SELECT + new_events_indexes.event_id, + 0, + linked_stream.initial_stream_version + new_events_indexes.index, + stream.stream_id, + new_events_indexes.stream_version + FROM + new_events_indexes, linked_stream, stream ) -INSERT INTO "<%= schema %>".stream_events -( - event_id, - stream_id, - stream_version, - original_stream_id, - original_stream_version -) -SELECT - new_events_indexes.event_id, - 0, - linked_stream.initial_stream_version + new_events_indexes.index, - stream.stream_id, - new_events_indexes.stream_version -FROM - new_events_indexes, linked_stream, stream; + +SELECT stream_id from stream; diff --git a/lib/event_store/sql/statements/insert_events_any_version.sql.eex b/lib/event_store/sql/statements/insert_events_any_version.sql.eex index 7be918e9..cfc250a0 100644 --- a/lib/event_store/sql/statements/insert_events_any_version.sql.eex +++ b/lib/event_store/sql/statements/insert_events_any_version.sql.eex @@ -57,20 +57,24 @@ WITH SET stream_version = stream_version + $2::bigint WHERE stream_id = 0 RETURNING stream_version - $2::bigint as initial_stream_version + ), + linked_stream_events AS ( + INSERT INTO "<%= schema %>".stream_events + ( + event_id, + stream_id, + stream_version, + original_stream_id, + original_stream_version + ) + SELECT + new_events_indexes.event_id, + 0, + linked_stream.initial_stream_version + new_events_indexes.index, + stream.stream_id, + stream.initial_stream_version + new_events_indexes.index + FROM + new_events_indexes, linked_stream, stream ) -INSERT INTO "<%= schema %>".stream_events -( - event_id, - stream_id, - stream_version, - original_stream_id, - original_stream_version -) -SELECT - new_events_indexes.event_id, - 0, - linked_stream.initial_stream_version + new_events_indexes.index, - stream.stream_id, - stream.initial_stream_version + new_events_indexes.index -FROM - new_events_indexes, linked_stream, stream; + +SELECT stream_id from stream; diff --git a/lib/event_store/storage/appender.ex b/lib/event_store/storage/appender.ex index ffd6b5c2..c88c740b 100644 --- a/lib/event_store/storage/appender.ex +++ b/lib/event_store/storage/appender.ex @@ -20,17 +20,19 @@ defmodule EventStore.Storage.Appender do events |> Stream.map(&encode_uuids/1) |> Stream.chunk_every(1_000) - |> Enum.each(fn batch -> + |> Enum.reduce(stream_id, fn batch, stream_id -> event_count = length(batch) - with :ok <- insert_event_batch(conn, stream_id, stream_uuid, batch, event_count, opts) do + with {:ok, new_stream_id} <- + insert_event_batch(conn, stream_id, stream_uuid, batch, event_count, opts) do Logger.debug("Appended #{event_count} event(s) to stream #{inspect(stream_uuid)}") - - :ok + new_stream_id else {:error, error} -> throw({:error, error}) end end) + + :ok catch {:error, error} = reply -> Logger.warning( @@ -110,9 +112,14 @@ defmodule EventStore.Storage.Appender do params = [stream_id_or_uuid, event_count] ++ build_insert_parameters(events) case Postgrex.query(conn, statement, params, opts) do - {:ok, %Postgrex.Result{num_rows: 0}} -> {:error, :not_found} - {:ok, %Postgrex.Result{}} -> :ok - {:error, error} -> handle_error(error) + {:ok, %Postgrex.Result{num_rows: 0}} -> + {:error, :not_found} + + {:ok, %Postgrex.Result{rows: [[stream_id]]}} -> + {:ok, stream_id} + + {:error, error} -> + handle_error(error) end end