fix(StreamEngine): graceful shutdown must wait for all events to be p…

[marcosschroh]

…rocessed before Streams are stopped. Related to #162

[github-actions]

PR Preview Action v1.4.7
Preview removed because the pull request was closed.
2024-02-08 11:59 UTC

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (ca70959) 95.75% compared to head (e013ae5) 95.53%.

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #171      +/-   ##
==========================================
- Coverage   95.75%   95.53%   -0.22%     
==========================================
  Files          23       23              
  Lines         849      852       +3     
==========================================
+ Hits          813      814       +1     
- Misses         36       38       +2     

Flag	Coverage Δ
unittests	95.53% <100.00%> (-0.22%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[marcosschroh]

This PR is addressing the first part of the issue #162 . With this new code, when a Stream is stopped (either directly or indirectly by the engine) the following thing will happen:

1. It will wait until the event in the chain (middlewares chain) has been processed
2. More importantly no more events will arrive in the getone.
3. The asyncio.Task that runs will be cancelled only and only when there are no more events.

If the end user has a really slow coroutine to process events and await stream.stop() is called, then it does not matter, the stop will wait for the event to be processed. Example in this test

Auto commit vs Manual commit:

Manual commit: With this new code it does not matter where the user commits (either at the beginning or end of the coroutine that handles the event). The reason is that the coroutine will run completely regardless how slow it can be when the stream is stopped, then we make sure that the commits always occurs

Auto commit: We have 2 scenarios:

• Scenario 1: Similar to manual commit, we know that the coroutine that handles the event will finish before the stop. if await stream.stop() is called when an event is being processed, then for point 2 we know that no more events will arrive, then we are not losing events.

• Scenario 2: If an event arrives and await stream.stop() is called just before the event start being processed then there is a small chance to lose the event. This could happen when the the stop is really slow and the auto_commit made the commit.

[woile]

Maybe add in a FAQ that messages are not lost.

What happens when a stream is cancelled?
...

[marcosschroh]

Maybe add in a FAQ that messages are not lost.

What happens when a stream is cancelled?
...

Good idea. We should add the new FAQ page