From 4f142e54eaabb7f49f9a6f65bc425acac4fa4fc3 Mon Sep 17 00:00:00 2001
From: Golo Roden <golo.roden@thenativeweb.io>
Date: Tue, 24 Mar 2026 15:05:32 +0100
Subject: [PATCH 1/7] fix: Raise exceptions in stream error handling instead of
 returning invalid tuples

Stream.resource/3 only accepts {list, acc} or {:halt, acc} as return values.
Returning {:error, reason} was invalid and would crash the stream consumer
with an unclear error. Now raises the exception directly, which propagates
cleanly to the consumer and allows Stream.resource's cleanup function to
properly close the HTTP connection.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 lib/eventsourcingdb.ex | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/eventsourcingdb.ex b/lib/eventsourcingdb.ex
index 011bd81..895adad 100644
--- a/lib/eventsourcingdb.ex
+++ b/lib/eventsourcingdb.ex
@@ -609,7 +609,7 @@ defmodule EventSourcingDB do
             {[message], response}
 
           {:error, reason} ->
-            {:error, reason}
+            raise(reason)
 
           # handle heartbeat case
           nil ->
@@ -617,7 +617,7 @@ defmodule EventSourcingDB do
         end
 
       {:error, reason} ->
-        {:error, reason}
+        raise(reason)
 
       # This is returned when the stream is done.
       {:ok, [:done]} ->

From 3e92ad3d281c96dbc24128abc13ac267bf5d66c5 Mon Sep 17 00:00:00 2001
From: Golo Roden <golo.roden@thenativeweb.io>
Date: Tue, 24 Mar 2026 15:05:55 +0100
Subject: [PATCH 2/7] fix: Correct typespec for write_events to return list of
 events

write_events returns a list of Event structs, not a single Event.
The typespec now correctly reflects response([Event.t()]).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 lib/eventsourcingdb.ex | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/eventsourcingdb.ex b/lib/eventsourcingdb.ex
index 895adad..25aefe1 100644
--- a/lib/eventsourcingdb.ex
+++ b/lib/eventsourcingdb.ex
@@ -200,13 +200,13 @@ defmodule EventSourcingDB do
 
   """
   @spec write_events(Client.t(), nonempty_list(EventCandidate.t()), [precondition()]) ::
-          response(Event.t())
+          response([Event.t()])
   def write_events(client, events, preconditions \\ []) when is_list(events) do
     request_one_shot(client, WriteEvents.new(events, preconditions))
   end
 
   @spec write_events!(Client.t(), nonempty_list(EventCandidate.t()), [precondition()]) ::
-          response!(Event.t())
+          response!([Event.t()])
   def write_events!(client, events, preconditions \\ []) when is_list(events) do
     request_one_shot!(client, WriteEvents.new(events, preconditions))
   end

From f06bd2b4ad586d2d48346c1bf16f248d9e987c76 Mon Sep 17 00:00:00 2001
From: Golo Roden <golo.roden@thenativeweb.io>
Date: Tue, 24 Mar 2026 15:06:17 +0100
Subject: [PATCH 3/7] fix: Default recursive option to false instead of
 enforcing it

Aligns with Go client behavior where Recursive defaults to false
(zero value). Users no longer need to explicitly set recursive
when creating options structs.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 lib/eventsourcingdb/request_options/observe_events.ex | 2 +-
 lib/eventsourcingdb/request_options/read_events.ex    | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/eventsourcingdb/request_options/observe_events.ex b/lib/eventsourcingdb/request_options/observe_events.ex
index 2a24c67..4d19e2f 100644
--- a/lib/eventsourcingdb/request_options/observe_events.ex
+++ b/lib/eventsourcingdb/request_options/observe_events.ex
@@ -5,7 +5,7 @@ defmodule EventSourcingDB.ObserveEventsOptions do
   use TypedStruct
 
   typedstruct do
-    field :recursive, boolean(), enforce: true
+    field :recursive, boolean(), default: false
     field :from_latest_event, EventSourcingDB.FromLatestEventOptions.t()
     field :lower_bound, EventSourcingDB.BoundOptions.t()
   end
diff --git a/lib/eventsourcingdb/request_options/read_events.ex b/lib/eventsourcingdb/request_options/read_events.ex
index a7386b5..f224ffb 100644
--- a/lib/eventsourcingdb/request_options/read_events.ex
+++ b/lib/eventsourcingdb/request_options/read_events.ex
@@ -5,7 +5,7 @@ defmodule EventSourcingDB.ReadEventsOptions do
   use TypedStruct
 
   typedstruct do
-    field :recursive, boolean(), enforce: true
+    field :recursive, boolean(), default: false
     field :order, :chronological | :antichronological
     field :from_latest_event, EventSourcingDB.FromLatestEventOptions.t()
     field :lower_bound, EventSourcingDB.BoundOptions.t()

From 340acc272a73146e3cc069ad31ea6afff8d18209 Mon Sep 17 00:00:00 2001
From: Golo Roden <golo.roden@thenativeweb.io>
Date: Tue, 24 Mar 2026 15:09:28 +0100
Subject: [PATCH 4/7] fix: Split FromLatestEventOptions into separate Read and
 Observe types

Introduces ReadFromLatestEventOptions (read_everything | read_nothing)
and ObserveFromLatestEventOptions (read_everything | wait_for_event)
to prevent invalid combinations at the type level. Aligns with Go client
which uses separate ReadFromLatestEvent and ObserveFromLatestEvent types.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 README.md                                     |  4 +-
 lib/eventsourcingdb.ex                        |  4 +-
 .../request_options/observe_events.ex         |  2 +-
 .../observe_from_latest_event.ex              | 40 +++++++++++++++++++
 .../request_options/read_events.ex            |  2 +-
 ...est_event.ex => read_from_latest_event.ex} |  9 ++---
 mix.exs                                       |  3 +-
 test/observe_events_test.exs                  |  4 +-
 test/read_events_test.exs                     |  6 +--
 9 files changed, 57 insertions(+), 17 deletions(-)
 create mode 100644 lib/eventsourcingdb/request_options/observe_from_latest_event.ex
 rename lib/eventsourcingdb/request_options/{from_latest_event.ex => read_from_latest_event.ex} (76%)

diff --git a/README.md b/README.md
index ea7cb8e..3210044 100644
--- a/README.md
+++ b/README.md
@@ -212,7 +212,7 @@ EventSourcingDB.read_events(
   "/books/42",
   %EventSourcingDB.ReadEventsOptions{
     recursive: false,
-    from_latest_event: %EventSourcingDB.FromLatestEventOptions{
+    from_latest_event: %EventSourcingDB.ReadFromLatestEventOptions{
       subject: "/books/42",
       type: "io.eventsourcingdb.library.book-borrowed",
       if_event_is_missing: :read_everything
@@ -297,7 +297,7 @@ EventSourcingDB.observe_events(
   "/books/42",
   %EventSourcingDB.ObserveEventsOptions{
     recursive: false,
-    from_latest_event: %EventSourcingDB.FromLatestEventOptions{
+    from_latest_event: %EventSourcingDB.ObserveFromLatestEventOptions{
       subject: "/books/42",
       type: "io.eventsourcingdb.library.book-borrowed",
       if_event_is_missing: :read_everything
diff --git a/lib/eventsourcingdb.ex b/lib/eventsourcingdb.ex
index 25aefe1..08dfc2f 100644
--- a/lib/eventsourcingdb.ex
+++ b/lib/eventsourcingdb.ex
@@ -292,7 +292,7 @@ defmodule EventSourcingDB do
     "/books/42",
     %EventSourcingDB.ReadEventsOptions{
       recursive: false,
-      from_latest_event: %EventSourcingDB.FromLatestEventOptions{
+      from_latest_event: %EventSourcingDB.ReadFromLatestEventOptions{
         subject: "/books/42",
         type: "io.eventsourcingdb.library.book-borrowed",
         if_event_is_missing: :read_everything
@@ -378,7 +378,7 @@ defmodule EventSourcingDB do
     "/books/42",
     %EventSourcingDB.ObserveEventsOptions{
       recursive: false,
-      from_latest_event: %EventSourcingDB.FromLatestEventOptions{
+      from_latest_event: %EventSourcingDB.ObserveFromLatestEventOptions{
         subject: "/books/42",
         type: "io.eventsourcingdb.library.book-borrowed",
         if_event_is_missing: :read_everything
diff --git a/lib/eventsourcingdb/request_options/observe_events.ex b/lib/eventsourcingdb/request_options/observe_events.ex
index 4d19e2f..68318f9 100644
--- a/lib/eventsourcingdb/request_options/observe_events.ex
+++ b/lib/eventsourcingdb/request_options/observe_events.ex
@@ -6,7 +6,7 @@ defmodule EventSourcingDB.ObserveEventsOptions do
 
   typedstruct do
     field :recursive, boolean(), default: false
-    field :from_latest_event, EventSourcingDB.FromLatestEventOptions.t()
+    field :from_latest_event, EventSourcingDB.ObserveFromLatestEventOptions.t()
     field :lower_bound, EventSourcingDB.BoundOptions.t()
   end
 
diff --git a/lib/eventsourcingdb/request_options/observe_from_latest_event.ex b/lib/eventsourcingdb/request_options/observe_from_latest_event.ex
new file mode 100644
index 0000000..fe1cd4e
--- /dev/null
+++ b/lib/eventsourcingdb/request_options/observe_from_latest_event.ex
@@ -0,0 +1,40 @@
+defmodule EventSourcingDB.ObserveFromLatestEventOptions do
+  @moduledoc """
+  Options for observing events starting from the latest event of a given type.
+  """
+  use TypedStruct
+
+  typedstruct enforce: true do
+    field :if_event_is_missing, :read_everything | :wait_for_event
+    field :subject, String.t()
+    field :type, String.t()
+  end
+
+  @spec new(keyword()) :: t()
+  def new(options) do
+    options =
+      options
+      |> Keyword.validate!([:if_event_is_missing, :subject, :type])
+
+    struct!(__MODULE__, options)
+  end
+
+  defimpl Jason.Encoder do
+    @spec encode(EventSourcingDB.ObserveFromLatestEventOptions.t(), Jason.Encode.opts()) ::
+            iodata()
+    def encode(value, opts) do
+      Jason.Encode.map(
+        %{
+          "ifEventIsMissing" =>
+            case value.if_event_is_missing do
+              :read_everything -> "read-everything"
+              :wait_for_event -> "wait-for-event"
+            end,
+          "subject" => value.subject,
+          "type" => value.type
+        },
+        opts
+      )
+    end
+  end
+end
diff --git a/lib/eventsourcingdb/request_options/read_events.ex b/lib/eventsourcingdb/request_options/read_events.ex
index f224ffb..6af0e95 100644
--- a/lib/eventsourcingdb/request_options/read_events.ex
+++ b/lib/eventsourcingdb/request_options/read_events.ex
@@ -7,7 +7,7 @@ defmodule EventSourcingDB.ReadEventsOptions do
   typedstruct do
     field :recursive, boolean(), default: false
     field :order, :chronological | :antichronological
-    field :from_latest_event, EventSourcingDB.FromLatestEventOptions.t()
+    field :from_latest_event, EventSourcingDB.ReadFromLatestEventOptions.t()
     field :lower_bound, EventSourcingDB.BoundOptions.t()
     field :upper_bound, EventSourcingDB.BoundOptions.t()
   end
diff --git a/lib/eventsourcingdb/request_options/from_latest_event.ex b/lib/eventsourcingdb/request_options/read_from_latest_event.ex
similarity index 76%
rename from lib/eventsourcingdb/request_options/from_latest_event.ex
rename to lib/eventsourcingdb/request_options/read_from_latest_event.ex
index 0cd6083..9fcc8c6 100644
--- a/lib/eventsourcingdb/request_options/from_latest_event.ex
+++ b/lib/eventsourcingdb/request_options/read_from_latest_event.ex
@@ -1,11 +1,11 @@
-defmodule EventSourcingDB.FromLatestEventOptions do
+defmodule EventSourcingDB.ReadFromLatestEventOptions do
   @moduledoc """
-  Reading events from when a certain event occured.
+  Options for reading events starting from the latest event of a given type.
   """
   use TypedStruct
 
   typedstruct enforce: true do
-    field :if_event_is_missing, :read_everything | :read_nothing | :wait_for_event
+    field :if_event_is_missing, :read_everything | :read_nothing
     field :subject, String.t()
     field :type, String.t()
   end
@@ -20,7 +20,7 @@ defmodule EventSourcingDB.FromLatestEventOptions do
   end
 
   defimpl Jason.Encoder do
-    @spec encode(EventSourcingDB.FromLatestEventOptions.t(), Jason.Encode.opts()) ::
+    @spec encode(EventSourcingDB.ReadFromLatestEventOptions.t(), Jason.Encode.opts()) ::
             iodata()
     def encode(value, opts) do
       Jason.Encode.map(
@@ -29,7 +29,6 @@ defmodule EventSourcingDB.FromLatestEventOptions do
             case value.if_event_is_missing do
               :read_everything -> "read-everything"
               :read_nothing -> "read-nothing"
-              :wait_for_event -> "wait-for-event"
             end,
           "subject" => value.subject,
           "type" => value.type
diff --git a/mix.exs b/mix.exs
index 964d2d0..33071ef 100644
--- a/mix.exs
+++ b/mix.exs
@@ -70,7 +70,8 @@ defmodule EventSourcingDB.MixProject do
         ],
         "Request Options": [
           EventSourcingDB.BoundOptions,
-          EventSourcingDB.FromLatestEventOptions,
+          EventSourcingDB.ObserveFromLatestEventOptions,
+          EventSourcingDB.ReadFromLatestEventOptions,
           EventSourcingDB.ObserveEventsOptions,
           EventSourcingDB.ReadEventsOptions
         ],
diff --git a/test/observe_events_test.exs b/test/observe_events_test.exs
index 38b074c..b863313 100644
--- a/test/observe_events_test.exs
+++ b/test/observe_events_test.exs
@@ -2,7 +2,7 @@ defmodule EventSourcingDBTest.ObserveEvents do
   alias EventSourcingDB.TestContainer
   alias EventSourcingDB.ObserveEventsOptions
   alias EventSourcingDB.BoundOptions
-  alias EventSourcingDB.FromLatestEventOptions
+  alias EventSourcingDB.ObserveFromLatestEventOptions
   import EventSourcingDBTest.Utils
   use ExUnit.Case, async: true
 
@@ -70,7 +70,7 @@ defmodule EventSourcingDBTest.ObserveEvents do
     events =
       EventSourcingDB.observe_events!(client, "/", %ObserveEventsOptions{
         recursive: true,
-        from_latest_event: %FromLatestEventOptions{
+        from_latest_event: %ObserveFromLatestEventOptions{
           subject: "/test",
           type: "io.eventsourcingdb.test.bar",
           if_event_is_missing: :read_everything
diff --git a/test/read_events_test.exs b/test/read_events_test.exs
index 968983f..270aa4b 100644
--- a/test/read_events_test.exs
+++ b/test/read_events_test.exs
@@ -1,7 +1,7 @@
 defmodule EventSourcingDBTest.ReadEvents do
   alias EventSourcingDB.Errors.TransmissionError
   alias EventSourcingDB.Client
-  alias EventSourcingDB.FromLatestEventOptions
+  alias EventSourcingDB.ReadFromLatestEventOptions
   alias EventSourcingDB.BoundOptions
   alias EventSourcingDB.ReadEventsOptions
   alias EventSourcingDB.TestContainer
@@ -181,7 +181,7 @@ defmodule EventSourcingDBTest.ReadEvents do
 
     events =
       EventSourcingDB.read_events!(client, "/test", %ReadEventsOptions{
-        from_latest_event: %FromLatestEventOptions{
+        from_latest_event: %ReadFromLatestEventOptions{
           subject: "/",
           type: "io.eventsourcingdb.test.does-not-exist",
           if_event_is_missing: :read_nothing
@@ -204,7 +204,7 @@ defmodule EventSourcingDBTest.ReadEvents do
 
     events =
       EventSourcingDB.read_events!(client, "/test", %ReadEventsOptions{
-        from_latest_event: %FromLatestEventOptions{
+        from_latest_event: %ReadFromLatestEventOptions{
           subject: "/marker",
           type: "io.eventsourcingdb.test",
           if_event_is_missing: :read_nothing

From 772fc7e4e1732ba41d259d8931909b83f5b65bee Mon Sep 17 00:00:00 2001
From: Golo Roden <golo.roden@thenativeweb.io>
Date: Tue, 24 Mar 2026 15:11:05 +0100
Subject: [PATCH 5/7] fix: Rename Event fields to idiomatic Elixir snake_case

Renames predecessorhash to predecessor_hash, datacontenttype to
data_content_type, and specversion to spec_version. Adds key mapping
in Event.new/1 and ManagementEvent.new/1 to translate JSON field names
to snake_case atoms. Updates all test assertions accordingly.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 lib/eventsourcingdb/events/event.ex           | 23 +++++++++++++++----
 .../events/management_event.ex                | 18 ++++++++++++---
 test/support/utils.ex                         |  8 +++----
 3 files changed, 37 insertions(+), 12 deletions(-)

diff --git a/lib/eventsourcingdb/events/event.ex b/lib/eventsourcingdb/events/event.ex
index 03b0564..2fe6e03 100644
--- a/lib/eventsourcingdb/events/event.ex
+++ b/lib/eventsourcingdb/events/event.ex
@@ -15,15 +15,21 @@ defmodule EventSourcingDB.Event do
 
   @signature_prefix "esdb:signature:v1:"
 
+  @key_mapping %{
+    "predecessorhash" => :predecessor_hash,
+    "datacontenttype" => :data_content_type,
+    "specversion" => :spec_version
+  }
+
   typedstruct do
     field :data, any(), enforce: true
-    field :datacontenttype, String.t(), enforce: true
+    field :data_content_type, String.t(), enforce: true
     field :hash, String.t(), enforce: true
     field :id, String.t(), enforce: true
-    field :predecessorhash, String.t(), enforce: true
+    field :predecessor_hash, String.t(), enforce: true
     field :signature, String.t()
     field :source, String.t(), enforce: true
-    field :specversion, String.t(), enforce: true
+    field :spec_version, String.t(), enforce: true
     field :subject, String.t(), enforce: true
     field :time, String.t(), enforce: true
     field :traceparent, String.t()
@@ -33,7 +39,14 @@ defmodule EventSourcingDB.Event do
 
   @spec new(map()) :: Event.t()
   def new(value \\ %{}) do
-    struct!(__MODULE__, value |> Map.new(fn {k, v} -> {String.to_existing_atom(k), v} end))
+    struct!(
+      __MODULE__,
+      value
+      |> Map.new(fn {k, v} ->
+        key = Map.get(@key_mapping, k, String.to_existing_atom(k))
+        {key, v}
+      end)
+    )
   end
 
   @doc """
@@ -55,7 +68,7 @@ defmodule EventSourcingDB.Event do
   @spec verify_hash(Event.t()) :: :ok | {:error, HashVerificationFailed.t()}
   def verify_hash(event) do
     metadata =
-      "#{event.specversion}|#{event.id}|#{event.predecessorhash}|#{event.time}|#{event.source}|#{event.subject}|#{event.type}|#{event.datacontenttype}"
+      "#{event.spec_version}|#{event.id}|#{event.predecessor_hash}|#{event.time}|#{event.source}|#{event.subject}|#{event.type}|#{event.data_content_type}"
 
     metadata_hash = :crypto.hash(:sha256, metadata) |> Base.encode16(case: :lower)
     data_hash = :crypto.hash(:sha256, Jason.encode!(event.data)) |> Base.encode16(case: :lower)
diff --git a/lib/eventsourcingdb/events/management_event.ex b/lib/eventsourcingdb/events/management_event.ex
index c94f620..a9f9712 100644
--- a/lib/eventsourcingdb/events/management_event.ex
+++ b/lib/eventsourcingdb/events/management_event.ex
@@ -5,12 +5,17 @@ defmodule EventSourcingDB.ManagementEvent do
   alias EventSourcingDB.ManagementEvent
   use TypedStruct
 
+  @key_mapping %{
+    "datacontenttype" => :data_content_type,
+    "specversion" => :spec_version
+  }
+
   typedstruct enforce: true do
     field :data, any()
-    field :datacontenttype, String.t()
+    field :data_content_type, String.t()
     field :id, String.t()
     field :source, String.t()
-    field :specversion, String.t()
+    field :spec_version, String.t()
     field :subject, String.t()
     field :time, String.t()
     field :type, String.t()
@@ -18,6 +23,13 @@ defmodule EventSourcingDB.ManagementEvent do
 
   @spec new(map()) :: ManagementEvent.t()
   def new(value \\ %{}) do
-    struct!(__MODULE__, value |> Map.new(fn {k, v} -> {String.to_existing_atom(k), v} end))
+    struct!(
+      __MODULE__,
+      value
+      |> Map.new(fn {k, v} ->
+        key = Map.get(@key_mapping, k, String.to_existing_atom(k))
+        {key, v}
+      end)
+    )
   end
 end
diff --git a/test/support/utils.ex b/test/support/utils.ex
index d36f7cf..fb742da 100644
--- a/test/support/utils.ex
+++ b/test/support/utils.ex
@@ -40,7 +40,7 @@ defmodule EventSourcingDBTest.Utils do
     assert_event_match_eventcandidate(event, candidate)
 
     assert event.id == Integer.to_string(expected_id), "ID should match"
-    assert event.predecessorhash == previous_event_hash, "Previous hash should match"
+    assert event.predecessor_hash == previous_event_hash, "Previous hash should match"
   end
 
   @spec assert_event_match_eventcandidate(Event.t(), EventCandidate.t()) :: any()
@@ -52,13 +52,13 @@ defmodule EventSourcingDBTest.Utils do
     assert event.type == candidate.type, "Type mismatch"
 
     # check metadata
-    assert event.datacontenttype == "application/json",
+    assert event.data_content_type == "application/json",
            "Data content type should be application/json"
 
     assert String.length(event.hash) == 64, "Hash should be present"
     assert not is_nil(event.id), "ID should be present"
-    assert not is_nil(event.predecessorhash), "Previous hash should be present"
-    assert event.specversion == "1.0", "Spec version should be 1.0"
+    assert not is_nil(event.predecessor_hash), "Previous hash should be present"
+    assert event.spec_version == "1.0", "Spec version should be 1.0"
   end
 
   @spec assert_event_match_eventcandidates([Event.t()], [EventCandidate.t()]) :: any()

From f54d4c04a5d074f22e3ce631478cf0a17ec5d477 Mon Sep 17 00:00:00 2001
From: Golo Roden <golo.roden@thenativeweb.io>
Date: Tue, 24 Mar 2026 15:13:48 +0100
Subject: [PATCH 6/7] docs: Add aborting documentation for all streaming
 operations

Adds "Aborting" sections to README and @doc annotations for read_events,
observe_events, run_eventql_query, read_subjects, and read_event_types.
Documents how to stop consuming lazy streams using Enum.take/2, stopping
iteration, or terminating the consuming process. Aligns with Go client
which documents context cancellation for the same operations.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 README.md              | 40 ++++++++++++++++++++++++++++++++++++++++
 lib/eventsourcingdb.ex | 22 ++++++++++++++++++++++
 2 files changed, 62 insertions(+)

diff --git a/README.md b/README.md
index 3210044..27eebda 100644
--- a/README.md
+++ b/README.md
@@ -223,6 +223,18 @@ EventSourcingDB.read_events(
 
 *Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*
 
+### Aborting Reading
+
+Since `read_events` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of events, or simply stop iterating:
+
+```elixir
+# Take only the first 10 events
+{:ok, events} = EventSourcingDB.read_events(client, "/books/42")
+first_ten = Enum.take(events, 10)
+```
+
+If you are consuming the stream in a separate process, you can abort by terminating that process.
+
 ## Running EventQL Queries
 
 To run an EventQL query, call the `run_eventql_query` function and provide the query as argument. The function returns a stream:
@@ -236,6 +248,15 @@ end
 
 *Note that each row returned by the stream matches the projection specified in your query.*
 
+### Aborting a Query
+
+Since `run_eventql_query` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of rows, or simply stop iterating:
+
+```elixir
+{:ok, rows} = EventSourcingDB.run_eventql_query(client, "FROM e IN events PROJECT INTO e")
+first_five = Enum.take(rows, 5)
+```
+
 ## Observing Events
 
 To observe all events of a subject, call the `observe_events` function with the subject.
@@ -308,6 +329,17 @@ EventSourcingDB.observe_events(
 
 *Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*
 
+### Aborting Observing
+
+Since `observe_events` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of events, or simply stop iterating:
+
+```elixir
+{:ok, events} = EventSourcingDB.observe_events(client, "/books/42")
+first_event = Enum.take(events, 1)
+```
+
+If you are consuming the stream in a separate process, you can abort by terminating that process.
+
 ## Registering an Event Schema
 
 To register an event schema, call the `register_event_schema` function and hand over an event type and the desired schema:
@@ -350,6 +382,10 @@ If you only want to list subjects within a specific branch, provide the desired
 EventSourcingDB.read_subjects(client, "/books")
 ```
 
+### Aborting Listing
+
+Since `read_subjects` returns a lazy stream, you can stop consuming it at any time by simply stopping iteration or using `Enum.take/2`.
+
 ## Reading Event Types
 
 To list all event types, call the `read_event_types` function. The function returns a stream from which you can retrieve one event type at a time:
@@ -361,6 +397,10 @@ case EventSourcingDB.read_event_types(client) do
 end
 ```
 
+### Aborting Listing
+
+Since `read_event_types` returns a lazy stream, you can stop consuming it at any time by simply stopping iteration or using `Enum.take/2`.
+
 ## Reading a Specific Event Type
 
 To read a specific event type, call the `read_event_type` function with the event type as an argument. The function returns the detailed event type, which includes the schema:
diff --git a/lib/eventsourcingdb.ex b/lib/eventsourcingdb.ex
index 08dfc2f..f6b32ca 100644
--- a/lib/eventsourcingdb.ex
+++ b/lib/eventsourcingdb.ex
@@ -303,6 +303,10 @@ defmodule EventSourcingDB do
 
   *Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*
 
+  ### Aborting Reading
+
+  Since `read_events` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of events, or simply stop iterating. If you are consuming the stream in a separate process, you can abort by terminating that process.
+
   """
   @spec read_events(Client.t(), String.t(), ReadEventsOptions.t() | nil) ::
           stream_response(Event.t())
@@ -389,6 +393,10 @@ defmodule EventSourcingDB do
 
   *Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*
 
+  ### Aborting Observing
+
+  Since `observe_events` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of events, or simply stop iterating. If you are consuming the stream in a separate process, you can abort by terminating that process.
+
   """
   @spec observe_events(Client.t(), String.t(), ObserveEventsOptions.t() | nil) ::
           stream_response(Event.t())
@@ -416,6 +424,10 @@ defmodule EventSourcingDB do
 
   *Note that each row returned by the stream matches the projection specified in your query.*
 
+  ### Aborting a Query
+
+  Since `run_eventql_query` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of rows, or simply stop iterating.
+
   """
   @spec run_eventql_query(Client.t(), String.t()) :: stream_response(any())
   def run_eventql_query(client, query) do
@@ -479,6 +491,11 @@ defmodule EventSourcingDB do
   ```elixir
   EventSourcingDB.read_subjects(client, "/books")
   ```
+
+  ### Aborting Listing
+
+  Since `read_subjects` returns a lazy stream, you can stop consuming it at any time by simply stopping iteration or using `Enum.take/2`.
+
   """
   @spec read_subjects(Client.t(), String.t()) :: stream_response(String.t())
   def read_subjects(client, base_subject) do
@@ -523,6 +540,11 @@ defmodule EventSourcingDB do
     {:error, reason} -> # ...
   end
   ```
+
+  ### Aborting Listing
+
+  Since `read_event_types` returns a lazy stream, you can stop consuming it at any time by simply stopping iteration or using `Enum.take/2`.
+
   """
   @spec read_event_types(Client.t()) :: stream_response(EventType.t())
   def read_event_types(client) do

From 5849e51e282046341c14f2a36b7ccda2b04189e9 Mon Sep 17 00:00:00 2001
From: Golo Roden <golo.roden@thenativeweb.io>
Date: Tue, 24 Mar 2026 15:17:14 +0100
Subject: [PATCH 7/7] fix: Use Map.get_lazy to avoid eager evaluation of
 String.to_existing_atom

Map.get evaluates the default value eagerly, causing
String.to_existing_atom("datacontenttype") to fail even when the key
exists in the mapping. Switch to Map.get_lazy which only evaluates
the fallback when the key is not found.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 lib/eventsourcingdb/events/event.ex            | 2 +-
 lib/eventsourcingdb/events/management_event.ex | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/eventsourcingdb/events/event.ex b/lib/eventsourcingdb/events/event.ex
index 2fe6e03..810ac85 100644
--- a/lib/eventsourcingdb/events/event.ex
+++ b/lib/eventsourcingdb/events/event.ex
@@ -43,7 +43,7 @@ defmodule EventSourcingDB.Event do
       __MODULE__,
       value
       |> Map.new(fn {k, v} ->
-        key = Map.get(@key_mapping, k, String.to_existing_atom(k))
+        key = Map.get_lazy(@key_mapping, k, fn -> String.to_existing_atom(k) end)
         {key, v}
       end)
     )
diff --git a/lib/eventsourcingdb/events/management_event.ex b/lib/eventsourcingdb/events/management_event.ex
index a9f9712..037a731 100644
--- a/lib/eventsourcingdb/events/management_event.ex
+++ b/lib/eventsourcingdb/events/management_event.ex
@@ -27,7 +27,7 @@ defmodule EventSourcingDB.ManagementEvent do
       __MODULE__,
       value
       |> Map.new(fn {k, v} ->
-        key = Map.get(@key_mapping, k, String.to_existing_atom(k))
+        key = Map.get_lazy(@key_mapping, k, fn -> String.to_existing_atom(k) end)
         {key, v}
       end)
     )