fix: Keep a stream open for as long as it runs, rather than end it after five minutes.

README.md

@@ -480,6 +480,8 @@ except HeartbeatTimeoutError:
 
 #### Aborting Observing
 
+Observing has no time limit of its own and runs until you abort it.
+
 If you need to abort observing use `break` or `return` within the `async for` loop. However, this only works if there is currently an iteration going on.
 
 To abort observing independently of that, store the generator in a variable, and close it explicitly:

eventsourcingdb/client.py

@@ -155,6 +155,7 @@ async def read_events(
         response: Response = await self.__http_client.post(
             path='/api/v1/read-events',
             request_body=request_body,
+            is_stream=True,
         )
 
         async with response:
@@ -182,6 +183,7 @@ async def run_eventql_query(self, query: str) -> AsyncGenerator[Any]:
         response: Response = await self.__http_client.post(
             path='/api/v1/run-eventql-query',
             request_body=request_body,
+            is_stream=True,
         )
 
         async with response:
@@ -219,6 +221,7 @@ async def observe_events(
         response: Response = await self.http_client.post(
             path='/api/v1/observe-events',
             request_body=request_body,
+            is_stream=True,
         )
 
         async with response:
@@ -272,6 +275,7 @@ async def read_subjects(
         response: Response = await self.http_client.post(
             path='/api/v1/read-subjects',
             request_body=request_body,
+            is_stream=True,
         )
 
         async with response:
@@ -330,6 +334,7 @@ async def read_event_types(self) -> EventTypeStream:
             response = await self.http_client.post(
                 path='/api/v1/read-event-types',
                 request_body='',
+                is_stream=True,
             )
         except CustomError:
             raise

eventsourcingdb/http_client/http_client.py

@@ -8,6 +8,11 @@
 from .get_post_headers import get_post_headers
 from .response import Response
 
+# A stream, such as observing events, runs for as long as the caller keeps
+# reading it, so only connecting is limited, not the request as a whole. Every
+# other request keeps the limit of the session.
+STREAM_TIMEOUT = aiohttp.ClientTimeout(total=None, sock_connect=30)
+
 
 class HttpClient:
     def __init__(
@@ -50,17 +55,24 @@ def join_segments(first: str, *rest: str) -> str:
 
         return f'{first_without_trailing_slash}/{rest_joined}'
 
-    async def post(self, path: str, request_body: str) -> Response:
+    async def post(
+        self,
+        path: str,
+        request_body: str,
+        is_stream: bool = False,
+    ) -> Response:
         if self.__session is None:
             await self.__initialize()
 
         url_path = HttpClient.join_segments(self.__base_url, path)
         headers = get_post_headers(self.__api_token)
+        timeout = STREAM_TIMEOUT if is_stream else self.__session.timeout  # type: ignore
 
         async_response = await self.__session.post(  # type: ignore
             url_path,
             data=request_body,
             headers=headers,
+            timeout=timeout,
         )
 
         response = Response(async_response)

tests/conftest.py

@@ -1,11 +1,16 @@
 from typing import ClassVar
 
+import aiohttp
+import pytest
 import pytest_asyncio
 
 from eventsourcingdb import EventCandidate
 
 from .shared.database import Database
 
+SESSION_TIMEOUT_SECONDS = 1
+TEST_DEADLINE_SECONDS = 3
+
 
 @pytest_asyncio.fixture
 async def database() -> Database:
@@ -98,3 +103,15 @@ async def events_for_mocked_server(
             trace_state=None,
         )
     ]
+
+
+@pytest.fixture
+def short_session_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
+    # By default, aiohttp limits every request of a session to five minutes.
+    # Shortening that limit shows within seconds which requests it applies to.
+    # Only sessions created after this fixture has run pick up the new limit.
+    monkeypatch.setattr(
+        aiohttp.client,
+        "DEFAULT_TIMEOUT",
+        aiohttp.ClientTimeout(total=SESSION_TIMEOUT_SECONDS, sock_connect=30),
+    )

tests/shared/stream_server.py

@@ -12,9 +12,9 @@
 
 class StreamServer:
     """
-    A local HTTP server that stands in for EventSourcingDB. It answers
-    observe-events and run-eventql-query with whatever the given function
-    writes to the stream, so that tests can control every single line.
+    A local HTTP server that stands in for EventSourcingDB. It answers every
+    POST request, whatever its path, with whatever the given function writes
+    to the stream, so that tests can control every single line.
     """
 
     def __init__(self, write_stream: WriteStream) -> None:
@@ -25,8 +25,7 @@ def __init__(self, write_stream: WriteStream) -> None:
 
     async def __aenter__(self) -> Self:
         application = web.Application()
-        application.router.add_post('/api/v1/observe-events', self.__handle)
-        application.router.add_post('/api/v1/run-eventql-query', self.__handle)
+        application.router.add_post('/{path:.*}', self.__handle)
 
         # Cancelling the handler when the client disconnects is what lets
         # tests notice that the client closed the connection.

tests/test_observe_events.py

@@ -6,14 +6,15 @@
 from eventsourcingdb import (
     Bound,
     BoundType,
+    Client,
     EventCandidate,
     IfEventIsMissingDuringObserve,
     ObserveEventsOptions,
     ObserveFromLatestEvent,
     ServerError,
 )
 
-from .conftest import TestData
+from .conftest import TEST_DEADLINE_SECONDS, TestData
 from .shared.database import Database
 from .shared.event.assert_event import assert_event_equals
 
@@ -411,3 +412,19 @@ async def process_events():
         assert events_processed > events_to_process, (
             "Expected to process some events before cancellation"
         )
+
+    @staticmethod
+    @pytest.mark.asyncio
+    @pytest.mark.usefixtures("short_session_timeout")
+    async def test_keeps_observing_for_longer_than_the_session_timeout(
+        database: Database,
+    ) -> None:
+        # A client of its own, because a session picks up its timeout when it is
+        # created, which has to happen after the timeout was shortened.
+        async with Client(database.get_base_url(), database.get_api_token()) as client:
+            with pytest.raises(TimeoutError):
+                async with asyncio.timeout(TEST_DEADLINE_SECONDS) as deadline:
+                    async for _ in client.observe_events("/", ObserveEventsOptions(recursive=True)):
+                        pass
+
+        assert deadline.expired(), "Observing must only end at the deadline of the test."

tests/test_run_eventql_query.py

@@ -1,9 +1,13 @@
+import asyncio
+
 import pytest
-from aiohttp import ClientConnectorDNSError
+from aiohttp import ClientConnectorDNSError, web
 
 from eventsourcingdb import EventCandidate
 
+from .conftest import TEST_DEADLINE_SECONDS
 from .shared.database import Database
+from .shared.stream_server import StreamServer
 
 
 class TestRunEventQLQuery:
@@ -71,3 +75,22 @@ async def test_reads_all_rows_the_query_returns(database: Database) -> None:
         second_row = rows_read[1]
         assert second_row["id"] == TestRunEventQLQuery.SECOND_EVENT_ID
         assert second_row["data"]["value"] == TestRunEventQLQuery.SECOND_EVENT_VALUE
+
+    @staticmethod
+    @pytest.mark.asyncio
+    @pytest.mark.usefixtures("short_session_timeout")
+    async def test_keeps_running_a_query_for_longer_than_the_session_timeout() -> None:
+        async def write_stream(response: web.StreamResponse) -> None:
+            while True:
+                await response.write(b'{"type":"heartbeat","payload":{}}\n')
+                await asyncio.sleep(1)
+
+        async with StreamServer(write_stream) as server:
+            client = server.get_client()
+
+            with pytest.raises(TimeoutError):
+                async with asyncio.timeout(TEST_DEADLINE_SECONDS) as deadline:
+                    async for _ in client.run_eventql_query("FROM e IN events PROJECT INTO e"):
+                        pass
+
+        assert deadline.expired(), "The query must only end at the deadline of the test."

tests/test_write_events.py

@@ -1,5 +1,7 @@
+import asyncio
+
 import pytest
-from aiohttp import ClientConnectorDNSError
+from aiohttp import ClientConnectorDNSError, web
 
 from eventsourcingdb import (
     EventCandidate,
@@ -10,8 +12,9 @@
     ServerError,
 )
 
-from .conftest import TestData
+from .conftest import TEST_DEADLINE_SECONDS, TestData
 from .shared.database import Database
+from .shared.stream_server import StreamServer
 
 
 class TestWriteSubjects:
@@ -322,3 +325,32 @@ async def test_throws_error_if_event_does_not_match_schema(
                     ),
                 ]
             )
+
+    @staticmethod
+    @pytest.mark.asyncio
+    @pytest.mark.usefixtures("short_session_timeout")
+    async def test_throws_an_error_if_server_answers_slower_than_the_session_timeout(
+        test_data: TestData,
+    ) -> None:
+        async def write_stream(_: web.StreamResponse) -> None:
+            await asyncio.Event().wait()
+
+        async with StreamServer(write_stream) as server:
+            client = server.get_client()
+
+            with pytest.raises(TimeoutError):
+                async with asyncio.timeout(TEST_DEADLINE_SECONDS) as deadline:
+                    await client.write_events(
+                        [
+                            EventCandidate(
+                                source=test_data.TEST_SOURCE_STRING,
+                                subject="/",
+                                type="com.foo.bar",
+                                data={},
+                            )
+                        ]
+                    )
+
+        assert not deadline.expired(), (
+            "The session timeout must end the request before the deadline of the test does."
+        )