Conversation
EventStore includes its own implementation of UUID.
```
==> eventstore
Compiling 60 files (.ex)
warning: <%# is deprecated, use <%!-- or add a space between <% and # instead
│
1 │ <%#
│ ~
│
└─ lib/event_store/sql/statements/insert_events.sql.eex:1: (file)
warning: <%# is deprecated, use <%!-- or add a space between <% and # instead
│
24 │ <%#
│ ~
│
└─ lib/event_store/sql/statements/insert_events.sql.eex:24: (file)
```
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds a new subscription option Changes
Sequence Diagram(s)sequenceDiagram
participant Sub as Subscription (GenServer)
participant FSM as SubscriptionFsm (FSM)
participant State as SubscriptionState
participant Timer as Erlang Timer
participant Client as Subscriber
Sub->>FSM: notify_partition_subscriber(events)
FSM->>State: enqueue_partition_events(events)
State->>State: update partitions / queue_size
alt First event in partition
State->>Timer: start_timer(buffer_flush_after)
Timer->>Timer: schedule {:flush_buffer, partition_key}
end
Timer->>Sub: {:flush_buffer, partition_key}
Sub->>FSM: flush_buffer(partition_key)
FSM->>State: flush_partition_on_timeout(partition_key)
State->>Client: deliver buffered events
Client->>Sub: acknowledge(last_event)
Sub->>FSM: handle_ack
FSM->>State: remove_in_flight / maybe restart_timer
alt pending events remain
State->>Timer: restart_timer(buffer_flush_after)
else partition empty
State->>Timer: cancel_timer(partition_key)
end
rect rgba(255, 100, 0, 0.5)
note right of Timer: per-partition timers, restarted/cancelled by FSM/State
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
lib/event_store.ex (1)
1150-1155: Clarify default behaviour inbuffer_flush_afterdocsThe sentence “When set to 0 (default), no time-based flushing is performed and events are only sent when the buffer_size is reached” doesn’t quite match the actual behaviour: even with
buffer_flush_after: 0, events are dispatched immediately up to each subscriber’sbuffer_sizeas long as there is an available subscriber (your new tests rely on this).Suggest rephrasing along the lines of:
- - `buffer_flush_after` (milliseconds) used to ensure events are flushed - to the subscriber after a period of time even if the buffer size has not - been reached. This ensures events are delivered with bounded latency - during less busy periods. When set to 0 (default), no time-based - flushing is performed and events are only sent when the buffer_size is - reached. Each partition has its own independent timer. + - `buffer_flush_after` (milliseconds) used to ensure events are flushed + to the subscriber after a period of time even if the subscriber's + `buffer_size` has not been filled. This ensures events are delivered + with bounded latency during less busy periods. When set to `0` + (default), no time-based flushing is performed and events are delivered + purely according to `buffer_size`/back‑pressure rules. Each partition + has its own independent timer.test/subscriptions/subscription_buffer_flush_after_test.exs (1)
1-387: Comprehensive tests forbuffer_flush_afterbehaviourThis suite does a good job exercising the new timeout semantics: basic flush on timeout vs buffer size, per‑partition timers, timer cancellation on empty queues or unsubscribe, back‑pressure edge cases, and ensuring no duplicate deliveries or ordering violations. The helper subscriber and assertion utilities are straightforward and make the intent of each scenario clear.
One gap you might consider (not mandatory) is a test that drives a subscription through a
catching_upphase withbuffer_flush_after > 0and a sizable backlog, to validate timer behaviour while the FSM is in non‑subscribedstates.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📥 Commits
Reviewing files that changed from the base of the PR and between 0bf4f2e and 1bbffcb004929454088fdadc04334f6899d32645.
📒 Files selected for processing (6)
.tool-versions(1 hunks)lib/event_store.ex(2 hunks)lib/event_store/subscriptions/subscription.ex(1 hunks)lib/event_store/subscriptions/subscription_fsm.ex(6 hunks)lib/event_store/subscriptions/subscription_state.ex(2 hunks)test/subscriptions/subscription_buffer_flush_after_test.exs(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (1)
test/subscriptions/subscription_buffer_flush_after_test.exs (3)
lib/event_store.ex (1)
append_to_stream(305-311)lib/event_store/subscriptions/subscription.ex (2)
ack(74-78)unsubscribe(83-85)test/subscriptions/support/stream_subscription_test_case.ex (1)
start_subscriber(433-437)
🔇 Additional comments (8)
.tool-versions (1)
2-2: Erlang version pin looks consistent
erlang 26.2.1matches theotp-26Elixir toolchain; nothing to flag here.lib/event_store.ex (1)
237-249: Type option extension forbuffer_flush_afteris consistentAdding
{:buffer_flush_after, non_neg_integer()}topersistent_subscription_optionmatches howSubscriptionFsm.new/3consumes the option and uses0as “disabled”. No issues here.lib/event_store/subscriptions/subscription.ex (1)
140-150:{:flush_buffer, partition_key}handler is wired correctlyThe new
handle_info/2clause cleanly delegates flush handling toSubscriptionFsm.flush_buffer/1and reapplies the returned FSM state, mirroring the existing patterns for:checkpointand advisory lock messages. No issues here; remaining behaviour is governed by the FSM implementation.lib/event_store/subscriptions/subscription_state.ex (1)
7-38: Buffer timer state and reset logic look soundAdding
buffer_flush_afterandbuffer_timersto the subscription state, and extendingreset_event_tracking/1to cancel and clear all active buffer timers alongside queue and checkpoint tracking, is consistent with the new timeout feature and prevents orphan timers after disconnect/reset. The implementation is straightforward and safe.Also applies to: 40-57
lib/event_store/subscriptions/subscription_fsm.ex (4)
12-32: Option wiring forbuffer_flush_aftermatches type/APIUsing
opts[:buffer_flush_after] || 0innew/3cleanly aligns with the public option and default behaviour (0 = disabled). No issues here.
511-532: Partition-aware enqueue and timer start behaviour looks correctDetecting “new partition” vs existing queues and starting the flush timer only on first enqueue per partition avoids redundant timers and keeps
queue_sizeincrements consistent. This integrates well with the later timer cancellation when a partition queue becomes empty.
559-593: Good: cancel partition timer when its queue is drainedTracking
partition_emptiedand cancelling the partition timer when the queue becomes empty ensures you don’t keep timers around for idle partitions. This is key to preventing stray{:flush_buffer, _}from acting on old state.
792-846: Timer helpers correctly guard against redundant timers and handle empty partitionsThe helper trio:
maybe_start_partition_timer/2(no‑op whenbuffer_flush_afteris0or a timer already exists),cancel_partition_timer/2(cancel + delete),flush_partition_on_timeout/2(no‑op if partition queue is absent, otherwise delegate tonotify_partition_subscriber/2),is well‑structured and matches the semantics you exercise in the tests (no duplicate sends, no action when queues are empty, respect back‑pressure). Aside from the broader FSM coverage issue noted earlier, the local logic here looks solid.
1bbffcb to
c66f5ec
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
lib/event_store/subscriptions/subscription_fsm.ex (1)
322-328: LGTM - Addresses previous review concern.This catch-all handler prevents FSM crashes when timers fire during transitional states (catch-up, disconnected, etc.). It correctly clears the timer reference without attempting to flush, allowing the subscription to continue operating normally.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📥 Commits
Reviewing files that changed from the base of the PR and between 1bbffcb004929454088fdadc04334f6899d32645 and c66f5ec6c6b01b5cfd11b6867b921febe3f8dc89.
📒 Files selected for processing (6)
.tool-versions(1 hunks)lib/event_store.ex(2 hunks)lib/event_store/subscriptions/subscription.ex(1 hunks)lib/event_store/subscriptions/subscription_fsm.ex(7 hunks)lib/event_store/subscriptions/subscription_state.ex(2 hunks)test/subscriptions/subscription_buffer_flush_after_test.exs(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (3)
- .tool-versions
- lib/event_store.ex
- lib/event_store/subscriptions/subscription_state.ex
🧰 Additional context used
🧬 Code graph analysis (1)
test/subscriptions/subscription_buffer_flush_after_test.exs (3)
lib/event_store.ex (1)
append_to_stream(305-311)lib/event_store/subscriptions/subscription.ex (2)
ack(74-78)unsubscribe(83-85)test/subscriptions/support/stream_subscription_test_case.ex (1)
start_subscriber(433-437)
🔇 Additional comments (10)
lib/event_store/subscriptions/subscription.ex (1)
140-150: LGTM!The new
handle_infoclause for{:flush_buffer, partition_key}follows the established pattern used by other handlers in this module (e.g.,:checkpoint). The delegation toSubscriptionFsm.flush_buffer/2and state application viaapply_subscription_to_state/2is consistent with the existing design.test/subscriptions/subscription_buffer_flush_after_test.exs (4)
635-683: Well-structured test helpers.The helper functions are clean and follow good testing patterns. The recursive
receive_all_events/1with a 500ms timeout is appropriate for accumulating events in tests without blocking indefinitely.
1-76: Comprehensive test coverage for the buffer_flush_after feature.The test suite thoroughly covers the core timeout functionality including:
- Partial batch flush on timeout expiry
- Immediate flush when buffer_size is reached before timeout
- Disabled behavior when buffer_flush_after is 0
- Complete flush of all pending events on timeout
338-457: Good coverage for catch-up state edge cases.These tests validate the catch-all
flush_bufferhandler behavior when timers fire during catch-up and transitional states. This addresses the potential crash scenarios identified in earlier reviews and ensures timer references are properly cleared without disrupting subscription operation.
250-336: Thorough back-pressure and edge case testing.The tests properly verify:
- Events stay queued when subscriber is at capacity during timeout
- No duplicate events when timer fires after events are already sent
- Timer restarts correctly when events remain after partial flush
- Clean timer cancellation on subscription stop
lib/event_store/subscriptions/subscription_fsm.ex (5)
26-26: LGTM!The
buffer_flush_afteroption is correctly initialized with a default of 0 (disabled), consistent with the documented behavior.
183-191: LGTM!The
flush_bufferhandler in the subscribed state correctly clears the timer reference (since the timer already fired) and attempts to flush the partition. The state transition back to:subscribedis appropriate.
519-540: LGTM!The timer start logic correctly:
- Detects new partitions via
Map.has_key?check- Only starts a timer for the first event in a partition
- Preserves existing timers for partitions already being tracked
This ensures the flush timeout is measured from when the first event arrives, not reset on every subsequent event.
582-600: LGTM!The partition emptiness tracking and timer cancellation is correctly implemented. When a partition's queue becomes empty after sending an event, the associated timer is properly cancelled via
cancel_partition_timer, preventing unnecessary timer fires.
797-865: Well-designed timer management helpers.Good separation of concerns:
maybe_start_partition_timer: Guards against disabled feature and duplicate timerscancel_partition_timer: For active timers that should be stopped (partition emptied via normal send)clear_partition_timer: For timers that already fired (just cleanup the reference)flush_partition_on_timeout: Handles the flush attempt and timer restart logicThe
flush_partition_on_timeoutcorrectly restarts the timer when events remain after a flush attempt (e.g., subscriber not available), ensuring eventual delivery.
c66f5ec to
ec0f063
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
lib/event_store/subscriptions/subscription.ex (1)
140-150: Flush-buffer messages are correctly integrated with the FSMThe new
handle_info({:flush_buffer, partition_key}, state)follows the same pattern as otherhandle_infoclauses (delegate toSubscriptionFsm, thenapply_subscription_to_state/2), so timer-driven flushes are cleanly fed into the FSM without special casing.If you find yourself debugging buffer timing issues frequently, consider adding a
Logger.debug/1here (similar to:subscribe_to_stream/{:events, _}) to log which partition is being flushed; otherwise this is fine as-is.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📥 Commits
Reviewing files that changed from the base of the PR and between c66f5ec6c6b01b5cfd11b6867b921febe3f8dc89 and ec0f063.
📒 Files selected for processing (6)
.tool-versions(1 hunks)lib/event_store.ex(2 hunks)lib/event_store/subscriptions/subscription.ex(1 hunks)lib/event_store/subscriptions/subscription_fsm.ex(8 hunks)lib/event_store/subscriptions/subscription_state.ex(2 hunks)test/subscriptions/subscription_buffer_flush_after_test.exs(1 hunks)
✅ Files skipped from review due to trivial changes (1)
- .tool-versions
🚧 Files skipped from review as they are similar to previous changes (2)
- test/subscriptions/subscription_buffer_flush_after_test.exs
- lib/event_store.ex
🧰 Additional context used
🧬 Code graph analysis (1)
lib/event_store/subscriptions/subscription.ex (3)
lib/event_store/notifications/listener.ex (1)
handle_info(36-45)test/support/collecting_subscriber.ex (2)
handle_info(50-56)handle_info(58-64)test/support/subscriber.ex (2)
handle_info(24-29)handle_info(31-36)
🔇 Additional comments (7)
lib/event_store/subscriptions/subscription_state.ex (1)
27-28: Clean timer state reset avoids stale buffer flush timersAdding
buffer_flush_after/buffer_timersto the struct and cancelling + clearing allbuffer_timersinreset_event_tracking/1keeps the FSM data consistent and prevents stale per-partition timers from surviving reconnects or disconnects. This aligns well with the new timer helpers inSubscriptionFsm.Also applies to: 41-53
lib/event_store/subscriptions/subscription_fsm.ex (6)
12-33: Initialization ofbuffer_flush_afteris consistent and safeWiring
buffer_flush_afterfromoptsintoSubscriptionStateinnew/3keeps configuration local to the FSM and matches the struct default of0(disabled). Usingopts[:buffer_flush_after] || 0preserves explicit0while defaultingnilto disabled, which is the right trade-off.
184-191: Subscribed-stateflush_buffer/1correctly encapsulates timeout-driven partition flushingThe
flush_buffer(partition_key)handler in the:subscribedstate first clears the timer reference and then delegates toflush_partition_on_timeout/2, which:
- Attempts to send queued events for that partition via
notify_partition_subscriber/2.- Restarts the timer only if the partition still has pending events.
This gives a clean, per-partition “flush and re-arm” loop without risking duplicate sends or dangling timers.
195-205: Max-capacity handling now avoids stale timers while respecting back-pressureThe updated
:max_capacitylogic looks solid:
- After
ack/2, if the queue isn’t empty you callrestart_timers_for_pending_partitions/1, which will only start timers for partitions that don’t already have one (viamaybe_start_partition_timer/2).- The
flush_buffer/1handler in:max_capacityjust clears the timer for that partition, leaving events queued to be drained as capacity becomes available.This combination prevents timers from going stale in
:max_capacitywhile still ensuring time-based flushing resumes once acknowledgements free up space.Also applies to: 215-219
324-330: Catch-allflush_buffer/1handler protects transitional states from crashes and stale timersAdding a generic
defevent flush_buffer(partition_key), state: statethat simply clears the timer reference and keeps the FSM in its current state ensures:
- Timers firing during
:initial,:request_catch_up,:catching_up,:disconnected, etc. won’t raise undefined-event errors.buffer_timersstays in sync even when you don’t want to actively flush partitions in those states.This neatly resolves the transitional-state timer edge cases.
521-542: Partition-aware enqueue and dequeue keep timers and queues in syncThe changes in
enqueue_event/3andnotify_partition_subscriber/3work well together:
enqueue_event/3distinguishes new partitions (is_new_partition) and only starts a buffer timer the first time a partition appears, avoiding redundant timers.notify_partition_subscriber/3now:
- Tracks whether the partition queue became empty and deletes it from
partitionswhen so.- Uses
max(queue_size - 1, 0)defensively when decrementing.- Cancels the timer via
cancel_partition_timer/2when the partition is emptied, ensuring no stray timers survive after all events for that partition have been dispatched.This keeps
partitions,queue_size, andbuffer_timersaligned and prevents timers from lingering on empty partitions.Also applies to: 584-603
800-877: Timer helper suite gives a clear, single-responsibility lifecycle for per-partition flushesThe new helpers (
maybe_start_partition_timer/2,cancel_partition_timer/2,clear_partition_timer/2,restart_timers_for_pending_partitions/1,flush_partition_on_timeout/2) provide a clean abstraction:
maybe_start_partition_timer/2respects disabled (buffer_flush_after: 0) configs and enforces one timer per partition.cancel_partition_timer/2andclear_partition_timer/2separate “cancel future message” from “drop reference after it has fired”, which matches howProcess.send_after/3behaves.restart_timers_for_pending_partitions/1offers a simple way to recover timers aftermax_capacityacks without duplicating logic.flush_partition_on_timeout/2centralizes “flush partition then possibly re-arm timer” semantics and reuses the existingnotify_partition_subscriber/2path.Overall this is a solid, maintainable way to manage the new time-based flush behavior.
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
lib/event_store/subscriptions/subscription_fsm.ex (1)
116-122: Consider restarting timers after ack in catching_up state for consistency.The
catching_upstate ack handler callsack_events(which callsnotify_subscribers) but doesn't restart timers if events remain, unlike themax_capacitystate (line 203). If a timer fires and is cleared by the catch-all handler during catch-up, and events remain after an ack, those events will lack timers until catch-up completes.While this may be acceptable since catch-up is designed to send events as fast as possible, it creates an inconsistency with
max_capacityand could violate the bounded-latency guarantee during longer catch-up phases.Consider adding timer restart logic similar to
max_capacity:defevent ack(ack, subscriber), data: %SubscriptionState{} = data do with {:ok, data} <- ack_events(data, ack, subscriber) do - catch_up_from_stream(data) + # Restart timers for partitions with pending events (may have been cleared during catch-up) + data = if data.queue_size > 0, do: restart_timers_for_pending_partitions(data), else: data + catch_up_from_stream(data) else reply -> respond(reply) end endThis ensures consistent bounded-latency behavior across all states.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
lib/event_store.ex(2 hunks)lib/event_store/subscriptions/subscription.ex(2 hunks)lib/event_store/subscriptions/subscription_fsm.ex(8 hunks)lib/event_store/subscriptions/subscription_state.ex(2 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- lib/event_store.ex
🧰 Additional context used
🧬 Code graph analysis (2)
lib/event_store/subscriptions/subscription.ex (1)
lib/event_store/subscriptions/subscription_state.ex (1)
cancel_all_buffer_timers(57-63)
lib/event_store/subscriptions/subscription_state.ex (1)
lib/event_store/subscriptions/subscription_fsm.ex (1)
new(12-34)
🔇 Additional comments (9)
lib/event_store/subscriptions/subscription_state.ex (1)
27-28: LGTM! Timer state management is clean and well-integrated.The new
buffer_flush_afterandbuffer_timersfields integrate cleanly with the existing struct. Thecancel_all_buffer_timers/1function is correctly placed before resetting state inreset_event_tracking/1, preventing timer leaks. The implementation properly handles the case whereProcess.cancel_timer/1returnsfalse.Also applies to: 40-63
lib/event_store/subscriptions/subscription.ex (1)
140-150: LGTM! Timer handling follows established patterns.The
handle_infoclause for:flush_buffercorrectly delegates to the FSM and applies the updated state, consistent with other message handlers. The termination cleanup properly cancels all buffer timers before checkpointing, preventing timer leaks on shutdown.Also applies to: 267-278
lib/event_store/subscriptions/subscription_fsm.ex (7)
26-26: LGTM! Initialization follows existing option pattern.The
buffer_flush_afterfield is correctly initialized from options with a sensible default of 0.
184-191: LGTM! Subscribed state flush handling is correct.The flush logic properly clears the timer reference and flushes the partition, notifying subscribers while remaining in the subscribed state.
201-224: LGTM! Max capacity timer restart ensures bounded latency.The max_capacity state correctly handles the buffer flush and ack lifecycle:
flush_bufferjust clears the timer since events can't be sent at capacity- After an ack,
restart_timers_for_pending_partitionsre-arms timers for queued events- This ensures events are flushed with bounded latency once capacity becomes available
329-338: LGTM! Catch-all handler prevents stale timer references.The catch-all
flush_bufferhandler correctly clears timer references when events fire in transitional states, preventing stale entries inbuffer_timers.
534-549: LGTM! Timer start on first partition event is correct.The logic correctly detects new partitions (not in the map) and starts timers only for the first event. This is the right trigger point for the timer lifecycle.
592-610: LGTM! Timer cancellation when partition empties is correct.The code properly detects when a partition's queue becomes empty and cancels the timer at that point, completing the timer lifecycle.
808-899: LGTM! Timer management helpers are well-designed.The helper functions provide clean abstractions:
maybe_start_partition_timerwith proper guards (buffer_flush_after > 0, no existing timer)cancel_partition_timervsclear_partition_timerdistinction is correct (cancel for manual cleanup, clear for already-fired timers)restart_timers_for_pending_partitionscorrectly iterates all pending partitionsflush_partition_on_timeoutproperly handles partial flushes and restarts timers if events remain
PR SummaryMedium Risk Overview Implements per-partition flush timers inside the subscription FSM, including timer lifecycle management (start on first queued event per partition, cancel on partition empty/unsubscribe/terminate, restart as needed under Updates docs ( Written by Cursor Bugbot for commit f6767f2. This will update automatically on new commits. Configure here. |
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Fix all issues with AI agents
In `@lib/event_store/storage/database.ex`:
- Around line 142-143: The change in lib/event_store/storage/database.ex that
maps {:error, %{postgres: %{code: :invalid_catalog_name}}} -> :ok breaks tests
and differs from schema.ex (which returns {:error, :already_down}); revert or
align behavior: update the pattern in the database drop function to return
{:error, :already_down} (not :ok) for invalid_catalog_name, or alternatively
change schema.ex and the handlers in lib/event_store/tasks/drop.ex to accept
:ok—pick one approach and make all three places consistent (database.ex handler,
schema.ex handler, and tasks/drop.ex message logic).
In `@lib/event_store/subscriptions/subscription_fsm.ex`:
- Around line 366-371: The file defines two catch-all defevent flush_buffer
handlers; remove the duplicate unreachable one so only the first handler
remains. Specifically, delete the second catch-all defevent
flush_buffer(partition_key), data: %SubscriptionState{} = data, state: state do
... end (the duplicate that also calls clear_partition_timer/2 and next_state/2)
and keep the initial defevent flush_buffer that references SubscriptionState,
clear_partition_timer, and next_state.
In `@test/subscriptions/subscription_buffer_checkpoint_resume_test.exs`:
- Around line 105-116: The test is double-acking events because
collect_and_ack_events already calls Subscription.ack; remove the redundant
explicit Subscription.ack calls for subscription1 (the calls immediately after
batch1 and batch2) so that events are only acknowledged once; keep the
append_to_stream and collect_and_ack_events calls unchanged and rely on
collect_and_ack_events to perform the acks and return the batches.
In `@test/subscriptions/subscription_buffer_edge_cases_test.exs`:
- Around line 338-351: The assertion Enum.uniq(nums) ==
Enum.sort(Enum.uniq(nums)) is insufficient; update the "very large single append
(500 events)" test to explicitly verify the event_number sequence is contiguous
and ordered by replacing that line with a sequence check that uses nums (from
collect_and_ack_events) to assert it equals the full consecutive range from
Enum.min(nums) to Enum.max(nums) (or an explicit 1..500 if event numbering
starts at 1). Keep the existing length assertion and use functions referenced
here (collect_and_ack_events, append_to_stream, events, event_number) to locate
and replace the redundant check with the contiguous-range assertion.
In `@test/subscriptions/subscription_buffer_selector_completeness_test.exs`:
- Around line 146-156: The test uses an imprecise assertion "assert
length(events) in [6, 7, 8]" even though the comment and scenario (global
event_numbers 1..9 with selector event_number > 2) imply a deterministic 7
events; update the assertion to assert length(events) == 7 (replace the in-list
check) and keep the subsequent verification of event_number values (nums) as-is,
and adjust the nearby comment to state the deterministic expectation of 7 events
so the test is unambiguous (referencing the variables/events and the
length(events) assertion in this test).
🧹 Nitpick comments (18)
test/shared_connection_pool_test.exs (1)
156-156: Consider extracting timeout values to a module attribute for consistency.The 5000ms timeouts are reasonable for this complex test involving event store restarts. For improved maintainability, consider using a module attribute:
`@subscription_timeout` 5_000This makes it easier to tune timeouts across tests and documents the intent.
Also applies to: 168-168
test/subscriptions/concurrent_subscription_test.exs (1)
853-872: Consider prepending and reversing for better performance (optional).The
acc ++ eventspattern on line 865 is O(n) per iteration, which could be slow for large event counts. For test code this is usually acceptable, but if tests become slow, consider prepending and reversing at the end.♻️ Optional optimization
- collect_events_and_ack(subscription, acc ++ events, expected_count, buffer_size, new_timeout) + collect_events_and_ack(subscription, Enum.reverse(events) ++ acc, expected_count, buffer_size, new_timeout)Then reverse at the return point in line 845:
- acc + Enum.reverse(acc)TEST_COVERAGE_SUMMARY.md (1)
211-227: Add language specifier to code fence.The code fence at line 211 is missing a language specifier, which is flagged by markdownlint. While purely informational, adding a language helps with syntax highlighting.
📝 Suggested fix
-``` +```text Total Tests: 121test/subscriptions/subscription_buffer_concurrent_subscribers_test.exs (2)
207-224: Inconsistentcollect_and_ack_with_timeoutimplementation.This helper returns
accimmediately in theafterclause (line 222), while similar helpers in other test files (e.g.,subscription_buffer_catchup_mode_test.exslines 425-440) continue recursion with reduced timeout. This could cause tests to return prematurely if there's a brief delay between batches.♻️ Align with other test files
defp collect_and_ack_with_timeout(subscription_pid, acc, remaining_timeout) do start = System.monotonic_time(:millisecond) receive do {:events, events} -> :ok = Subscription.ack(subscription_pid, events) elapsed = System.monotonic_time(:millisecond) - start new_timeout = remaining_timeout - elapsed collect_and_ack_with_timeout(subscription_pid, acc ++ events, new_timeout) after min(remaining_timeout, 200) -> - acc + elapsed = System.monotonic_time(:millisecond) - start + new_timeout = remaining_timeout - elapsed + collect_and_ack_with_timeout(subscription_pid, acc, new_timeout) end end
170-186: Assertion may be too lenient.Line 185 asserts
length(events) >= 400when 500 events are appended (10 iterations × 50 events). This allows up to 100 events to be "lost" which seems inconsistent with the test description "without leaking resources." Consider asserting the exact count.📝 Suggested fix
- assert length(events) >= 400 + assert length(events) == 500test/subscriptions/subscription_buffer_flush_diagnostics_test.exs (1)
111-127: Hardcoded timeout decrement in collect_with_logging.Line 121 decrements timeout by a hardcoded 100ms instead of measuring actual elapsed time. This is acceptable for diagnostic purposes but note it's imprecise.
♻️ Optional: Use actual elapsed time
defp collect_with_logging(subscription_pid, acc, remaining_timeout: remaining) do + start = System.monotonic_time(:millisecond) receive do {:events, events} -> IO.puts("Received #{length(events)} events") Enum.each(events, &IO.inspect(&1.event_number, label: " event_number")) - collect_with_logging(subscription_pid, acc ++ events, remaining_timeout: remaining - 100) + elapsed = System.monotonic_time(:millisecond) - start + collect_with_logging(subscription_pid, acc ++ events, remaining_timeout: remaining - elapsed) after 200 -> IO.puts("No events received in 200ms") - collect_with_logging(subscription_pid, acc, remaining_timeout: remaining - 200) + elapsed = System.monotonic_time(:millisecond) - start + collect_with_logging(subscription_pid, acc, remaining_timeout: remaining - elapsed) end endtest/subscriptions/subscription_buffer_invariants_test.exs (2)
56-61: Clarify global event number assertion.The assertion at line 60 expects
event_nums == Enum.to_list(1..15), which implies events are delivered in strict global order. With partitioned subscriptions, events may be delivered out of global order (though still in order per-partition). Thecollect_and_ack_eventshelper collects all events and the assertion checks the sorted result matches the range, which is correct.However, the assertion message "Global event numbers should be [1..15]" could be misleading since it suggests strict ordering. Consider clarifying:
📝 Suggested clarification
# Verify global event number sequence event_nums = Enum.map(events, & &1.event_number) - assert event_nums == Enum.to_list(1..15), - "Global event numbers should be [1..15], got #{inspect(event_nums)}" + assert Enum.sort(event_nums) == Enum.to_list(1..15), + "All global event numbers [1..15] should be present, got #{inspect(event_nums)}"
195-201: O(n²) complexity in sequential check loop.The loop uses
Enum.at/2which is O(n) per call, making the entire loop O(n²). For test code with small n this is acceptable, but could be optimized if needed.♻️ Optional: Linear time alternative
- # Verify no duplicates in order - for i <- 0..(length(event_nums) - 2) do - curr = Enum.at(event_nums, i) - next = Enum.at(event_nums, i + 1) - - assert next == curr + 1, - "Event numbers should be sequential, got #{curr} then #{next}" - end + # Verify no duplicates and sequential + event_nums + |> Enum.chunk_every(2, 1, :discard) + |> Enum.each(fn [curr, next] -> + assert next == curr + 1, + "Event numbers should be sequential, got #{curr} then #{next}" + end)test/subscriptions/subscription_buffer_selector_completeness_test.exs (2)
375-387: Consider prepending and reversing for O(1) accumulation.Using
acc ++ eventsis O(n) where n is the accumulated list size. For large test scenarios, this could slow down tests. This pattern is duplicated across all test files.Suggested refactor
defp collect_and_ack_with_timeout(subscription_pid, acc, remaining_timeout) do start = System.monotonic_time(:millisecond) receive do {:events, events} -> :ok = Subscription.ack(subscription_pid, events) elapsed = System.monotonic_time(:millisecond) - start new_timeout = remaining_timeout - elapsed - collect_and_ack_with_timeout(subscription_pid, acc ++ events, new_timeout) + # Prepend reversed events for O(1) operation, reverse at end + collect_and_ack_with_timeout(subscription_pid, Enum.reverse(events) ++ acc, new_timeout) after min(remaining_timeout, 200) -> - acc + Enum.reverse(acc) end end
355-365: Helper functions are duplicated across all test files.The
subscribe_to_all_streams/1,append_to_stream/3, andcollect_and_ack_events/2helpers are nearly identical across all 6 test files in this PR. Consider extracting these into a shared test support module (e.g.,test/support/subscription_test_helpers.ex) to reduce duplication and simplify maintenance.test/subscriptions/subscription_buffer_flush_after_test.exs (1)
237-247: Concurrency test may have timing sensitivity.The test relies on receiving two events messages in sequence from different subscribers. In CI environments with varying load, the 500ms timeout might occasionally be insufficient, or messages could arrive in unexpected order.
Consider adding a small tolerance or using
assert_receivewith pattern matching that's order-independent:# Collect both events regardless of order events = for _ <- 1..2 do assert_receive {:events, events, _sub}, 500 events endtest/subscriptions/subscription_buffer_correctness_focus_test.exs (1)
179-180: Unconventional pattern for extracting events from assert_receive.Using
assert_receive({:events, _}, 500) |> elem(1)is unusual and less readable.assert_receivereturns the matched message, soelem(1)extracts the events. Consider using the standard pattern match approach for clarity.Suggested refactor
- batch1 = assert_receive({:events, _}, 500) |> elem(1) + assert_receive {:events, batch1}, 500 assert length(batch1) == 2test/subscriptions/subscription_buffer_large_scale_test.exs (1)
407-426: Performance assertion doesn't actually verify linearity.The test name claims "batch delivery time increases linearly with event count" but only asserts
total_time < 5000. This doesn't verify linearity - it only sets an upper bound. To test linearity, you'd need to compare times across different event counts.Consider renaming to reflect what's actually being tested, or enhancing the test:
Suggested rename
- test "batch delivery time increases linearly with event count" do + test "batch delivery completes within reasonable time bounds" dotest/subscriptions/subscription_buffer_edge_cases_test.exs (1)
203-256: Test logic is convoluted and hard to follow.This test attempts to verify single-event batches but uses a confusing pattern of building a
batcheslist with multiplereceiveblocks and inline acking at inconsistent points. The logic is error-prone and difficult to maintain.Suggested refactor for clarity
- test "single event per batch (buffer_size = 1)" do - {:ok, subscription} = - subscribe_to_all_streams(buffer_size: 1, buffer_flush_after: 50) - - append_to_stream("stream1", 3) - - # Expect 3 single-event batches - batches = [] - - batches = - (batches ++ - [ - receive do - {:events, b} -> b - after - 1000 -> [] - end - ]) - |> Enum.filter(&(length(&1) > 0)) - ... # continues with similar complex patterns + test "single event per batch (buffer_size = 1)" do + {:ok, subscription} = + subscribe_to_all_streams(buffer_size: 1, buffer_flush_after: 50) + + append_to_stream("stream1", 3) + + # Collect 3 single-event batches + batches = + for _ <- 1..3 do + assert_receive {:events, batch}, 1000 + assert length(batch) == 1, "Each batch should have exactly 1 event" + Subscription.ack(subscription, batch) + batch + end + + assert length(batches) == 3 + assert Enum.all?(batches, &(length(&1) == 1)) + endtest/subscriptions/subscription_buffer_checkpoint_resume_test.exs (1)
356-385: Missingsubscribe_to_all_streamshelper unlike other test files.This test file calls
EventStore.subscribe_to_all_streamsdirectly in each test rather than using a helper. While this works (and allows using the same subscription_name for resume tests), it's inconsistent with other test files. Consider adding the helper with an optionalnameparameter for consistency:defp subscribe_to_all_streams(name \\ UUID.uuid4(), opts) do {:ok, subscription} = EventStore.subscribe_to_all_streams(name, self(), opts) assert_receive {:subscribed, ^subscription} {:ok, subscription, name} endtest/subscriptions/subscription_buffer_catchup_mode_test.exs (1)
269-358: Consider refactoring repetitive receive blocks.This test contains 8 nearly identical receive blocks. While the intent is clear (careful ACKing to maintain back-pressure), this can be simplified using a loop or the existing
collect_and_ack_eventshelper.♻️ Suggested refactor using Enum.reduce
test "catch-up doesn't lose events during max_capacity" do {:ok, subscription} = subscribe_to_all_streams( buffer_size: 2, buffer_flush_after: 100 ) # Append 15 events append_to_stream("stream1", 15) # Collect with careful ACKing to maintain back-pressure - events = [] - - events = - receive do - {:events, b1} -> - Subscription.ack(subscription, b1) - events ++ b1 - after - 1000 -> events - end - - events = - receive do - {:events, b2} -> - Subscription.ack(subscription, b2) - events ++ b2 - after - 1000 -> events - end - - events = - receive do - {:events, b3} -> - Subscription.ack(subscription, b3) - events ++ b3 - after - 1000 -> events - end - - events = - receive do - {:events, b4} -> - Subscription.ack(subscription, b4) - events ++ b4 - after - 1000 -> events - end - - events = - receive do - {:events, b5} -> - Subscription.ack(subscription, b5) - events ++ b5 - after - 1000 -> events - end - - events = - receive do - {:events, b6} -> - Subscription.ack(subscription, b6) - events ++ b6 - after - 1000 -> events - end - - events = - receive do - {:events, b7} -> - Subscription.ack(subscription, b7) - events ++ b7 - after - 1000 -> events - end - - events = - receive do - {:events, b8} -> - Subscription.ack(subscription, b8) - events ++ b8 - after - 1000 -> events - end + # With buffer_size: 2 and 15 events, expect up to 8 batches + events = Enum.reduce(1..8, [], fn _i, acc -> + receive do + {:events, batch} -> + Subscription.ack(subscription, batch) + acc ++ batch + after + 1000 -> acc + end + end) # Verify all events received assert length(events) == 15 nums = Enum.map(events, & &1.event_number) assert nums == Enum.to_list(1..15) endtest/subscriptions/subscription_buffer_comprehensive_test.exs (2)
538-555: Behavior differs from the catch-up mode test file's helper.This
collect_and_ack_with_timeoutreturns immediately when no events are received within the polling window (line 553), while the catch-up mode test file's version continues recursing with reduced timeout. Both behaviors are valid for their respective test scenarios, but worth noting for maintainability.
557-562: Unused parameter in helper function.The
_subscription_pidparameter is not used inmeasure_collection/2. Consider removing it if not needed.♻️ Suggested fix
- defp measure_collection(_subscription_pid, fun) do + defp measure_collection(fun) do start = System.monotonic_time(:millisecond) result = fun.() elapsed = System.monotonic_time(:millisecond) - start {result, elapsed} endAnd update the call site at line 141:
- measure_collection(subscription, fn -> + measure_collection(fn -> collect_and_ack_events(subscription, timeout: 1500) end)
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
f9998ab to
e3cb06e
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In `@test/subscriptions/subscription_buffer_concurrent_subscribers_test.exs`:
- Around line 219-231: collect_and_ack_with_timeout/3 currently returns acc as
soon as the first idle window fires, which can drop late events; update the
after branch so instead of returning immediately it checks remaining_timeout
and, if still > 0, recursively calls
collect_and_ack_with_timeout(subscription_pid, acc, remaining_timeout -
min(remaining_timeout, 200)) (or equivalently compute the new remaining timeout)
to keep listening until the total timeout is exhausted, otherwise return acc;
keep the existing receive handling (Subscription.ack and acc ++ events) and use
the same min(remaining_timeout, 200) logic to step the timeout.
In `@test/subscriptions/subscription_buffer_invariants_test.exs`:
- Around line 391-417: The two tests ("last_received always >= last_sent" and
"checkpoint progress matches acked events") are only asserting event counts and
not the actual subscription invariants; update each test to read and assert the
subscription state instead of—or in addition to—counting events: after
collect_and_ack_events(subscription, ...), call the subscription state API
(e.g., get_subscription_state(subscription) or
read_subscription_checkpoint(subscription)) to assert that state.last_received
>= state.last_sent in the first test and that state.checkpoint (or
state.last_checkpointed_event_index) equals the number of acked events in the
second test; if no such API exists, remove or skip these tests to avoid false
confidence.
In `@test/subscriptions/subscription_buffer_large_scale_test.exs`:
- Around line 91-93: Update the incorrect explanatory comment above the
expected_total calculation: replace the manual sum "6*5 + 5*4 + 5*3 + 5*2 + 5*1
= 30+20+15+10+5 = 80" with the correct reasoning for rem(i, 5) + 1 over 1..30
(pattern 2,3,4,5,1 repeated 6 times -> 6 * 15 = 90), referencing the existing
expression expected_total = Enum.sum(Enum.map(1..30, fn i -> rem(i, 5) + 1 end))
so the comment accurately documents the computed value.
♻️ Duplicate comments (3)
test/subscriptions/subscription_buffer_selector_completeness_test.exs (1)
146-156: Imprecise assertion indicates unclear expected behavior.The assertion
assert length(events) in [6, 7, 8]suggests uncertainty. Based on the comment (lines 146-149), the selector filtersevent_number > 2across global event numbers 1-9, which should yield exactly 7 events (3, 4, 5, 6, 7, 8, 9).Proposed fix
- assert length(events) in [6, 7, 8] + assert length(events) == 7, + "Should receive 7 events (event_number 3-9, filtering out 1 and 2)"test/subscriptions/subscription_buffer_edge_cases_test.exs (1)
350-352: Sequence integrity assertion is ineffective (duplicate).
Enum.uniq(nums) == Enum.sort(Enum.uniq(nums))doesn’t detect gaps or ordering issues. Use a contiguous-range check instead.🛠️ Suggested fix
# Verify sequence integrity nums = Enum.map(events, & &1.event_number) - assert Enum.uniq(nums) == Enum.sort(Enum.uniq(nums)) + expected = Enum.to_list(Enum.min(nums)..Enum.max(nums)) + assert nums == expected, "Events should form a contiguous sequence"test/subscriptions/subscription_buffer_checkpoint_resume_test.exs (1)
107-117: Avoid double‑acking batches already acknowledged by collect_and_ack_events.
collect_and_ack_events/2already callsSubscription.ack/2, so the explicit acks here are redundant and can mask issues if acks aren’t idempotent.🧹 Suggested fix
append_to_stream("stream1", 3) batch1 = collect_and_ack_events(subscription1, timeout: 1000) - Subscription.ack(subscription1, batch1) # Wait for checkpoint to write Process.sleep(200) # Append more while still subscribed append_to_stream("stream1", 3, 3) batch2 = collect_and_ack_events(subscription1, timeout: 1000) - Subscription.ack(subscription1, batch2)
🧹 Nitpick comments (5)
test/subscriptions/subscription_buffer_flush_diagnostics_test.exs (1)
94-99: Dead assignment:all_events = []is immediately overwritten.The variable
all_eventsis assigned an empty list on line 94 but immediately reassigned on line 98.Proposed fix
- all_events = [] - start = System.monotonic_time(:millisecond) - - # Collect all events with timeout - all_events = - collect_with_logging(subscription, all_events, remaining_timeout: 2000) + start = System.monotonic_time(:millisecond) + + # Collect all events with timeout + all_events = + collect_with_logging(subscription, [], remaining_timeout: 2000)test/subscriptions/subscription_buffer_checkpoint_resume_test.exs (2)
209-233: Loosen the timeout assertion to reduce CI jitter.The fixed 250ms bound is tight; tying it to
buffer_flush_afterkeeps the intent while being less flaky under load.♻️ Suggested adjustment
- {:ok, subscription} = - EventStore.subscribe_to_all_streams( - subscription_name, - self(), - buffer_size: 10, - buffer_flush_after: 100, - checkpoint_after: 500 - ) + buffer_flush_after = 100 + {:ok, subscription} = + EventStore.subscribe_to_all_streams( + subscription_name, + self(), + buffer_size: 10, + buffer_flush_after: buffer_flush_after, + checkpoint_after: 500 + ) @@ - assert elapsed < 250, "Should flush via timeout, not wait for checkpoint" + max_latency = buffer_flush_after * 3 + assert elapsed < max_latency, "Should flush via timeout, not wait for checkpoint"
374-385: Wait out the full remaining timeout in collect_and_ack_with_timeout.Returning after a 200ms quiet period ignores
remaining_timeoutand can truncate collection when batches arrive slightly later.♻️ Suggested adjustment
after min(remaining_timeout, 200) -> - acc + elapsed = System.monotonic_time(:millisecond) - start + new_timeout = remaining_timeout - elapsed + collect_and_ack_with_timeout(subscription_pid, acc, new_timeout) endtest/subscriptions/subscription_buffer_comprehensive_test.exs (2)
98-151: Parameterize latency bounds to reduce flaky timing.Hard-coded limits (e.g., 200ms/1000ms) can be brittle under CI load. Deriving bounds from
buffer_flush_afterpreserves intent with more slack.♻️ Suggested adjustment
- {:ok, subscription} = - subscribe_to_all_streams(buffer_size: 10, buffer_flush_after: 80) + buffer_flush_after = 80 + {:ok, subscription} = + subscribe_to_all_streams(buffer_size: 10, buffer_flush_after: buffer_flush_after) @@ - assert elapsed < 200, "Events should be delivered within bounded latency, took #{elapsed}ms" + max_latency = buffer_flush_after * 3 + assert elapsed < max_latency, + "Events should be delivered within bounded latency, took #{elapsed}ms" @@ - {:ok, subscription} = - subscribe_to_all_streams(buffer_size: 20, buffer_flush_after: 100) + buffer_flush_after = 100 + {:ok, subscription} = + subscribe_to_all_streams(buffer_size: 20, buffer_flush_after: buffer_flush_after) @@ - assert Enum.all?(timings, &(&1 < 200)), + max_latency = buffer_flush_after * 3 + assert Enum.all?(timings, &(&1 < max_latency)), "All cycles should maintain latency bounds, got: #{inspect(timings)}" @@ - {:ok, subscription} = - subscribe_to_all_streams(buffer_size: 2, buffer_flush_after: 100) + buffer_flush_after = 100 + {:ok, subscription} = + subscribe_to_all_streams(buffer_size: 2, buffer_flush_after: buffer_flush_after) @@ - assert total_time < 1000, + max_total_time = buffer_flush_after * 10 + assert total_time < max_total_time, "Back-pressure shouldn't prevent bounded latency, took #{total_time}ms"
549-561: Honor remaining_timeout in collect_and_ack_with_timeout.Returning after a 200ms idle period disregards
remaining_timeoutand can miss late batches.♻️ Suggested adjustment
after min(remaining_timeout, 200) -> - acc + elapsed = System.monotonic_time(:millisecond) - start + new_timeout = remaining_timeout - elapsed + collect_and_ack_with_timeout(subscription_pid, acc, new_timeout) end
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In `@test/subscriptions/subscription_buffer_checkpoint_resume_test.exs`:
- Around line 283-291: Replace the loose check that all event_number values are
> 2 with a precise equality check: after collecting events2 (via
collect_and_ack_events) compute nums = Enum.map(events2, & &1.event_number) and
assert Enum.sort(nums) == Enum.to_list(3..8) (or pattern-match the exact list)
so you verify you received exactly events 3 through 8 and not a replay of
earlier items.
- Around line 210-231: The timing assertion is brittle; update the assertion
that the buffer flush occurred before the checkpoint by comparing elapsed
against the configured checkpoint_after (or checkpoint_after minus a small
slack), not a hardcoded 250ms. Locate the subscribe call using
EventStore.subscribe_to_all_streams with options buffer_flush_after: 100 and
checkpoint_after: 500 and replace assert elapsed < 250 with an assertion like
assert elapsed < checkpoint_after (or assert elapsed < checkpoint_after -
slack_ms) so the test compares to the actual configured checkpoint threshold
rather than an arbitrary fixed value.
In `@test/subscriptions/subscription_buffer_selector_completeness_test.exs`:
- Around line 371-388: The after branch in collect_and_ack_with_timeout returns
immediately on a short wait slice, truncating collection; instead compute wait =
min(remaining_timeout, 200), measure elapsed time in the after branch (using
System.monotonic_time), subtract it from remaining_timeout to get new_timeout,
and recurse (call collect_and_ack_with_timeout(subscription_pid, acc,
new_timeout)) unless new_timeout <= 0; keep the existing receive handling that
acks via Subscription.ack and appends events to acc so the loop continues until
the overall timeout elapses.
🧹 Nitpick comments (1)
test/subscriptions/subscription_buffer_edge_cases_test.exs (1)
400-421: Consider making the idle wait configurable to avoid premature truncation.
min(remaining_timeout, 200)can stop collection early if batches are delayed >200ms, which risks flakiness on slow CI. Making the idle cap configurable preserves responsiveness while avoiding accidental early exits.♻️ Suggested refactor (configurable idle timeout)
-defp collect_and_ack_events(subscription_pid, timeout: timeout) do - collect_and_ack_with_timeout(subscription_pid, [], timeout) +defp collect_and_ack_events(subscription_pid, timeout: timeout, idle_timeout: idle_timeout \\ 200) do + collect_and_ack_with_timeout(subscription_pid, [], timeout, idle_timeout) end -defp collect_and_ack_with_timeout(_subscription_pid, acc, remaining_timeout) +defp collect_and_ack_with_timeout(_subscription_pid, acc, remaining_timeout, _idle_timeout) when remaining_timeout <= 0 do acc end -defp collect_and_ack_with_timeout(subscription_pid, acc, remaining_timeout) do +defp collect_and_ack_with_timeout(subscription_pid, acc, remaining_timeout, idle_timeout) do start = System.monotonic_time(:millisecond) receive do {:events, events} -> :ok = Subscription.ack(subscription_pid, events) elapsed = System.monotonic_time(:millisecond) - start new_timeout = remaining_timeout - elapsed - collect_and_ack_with_timeout(subscription_pid, acc ++ events, new_timeout) + collect_and_ack_with_timeout(subscription_pid, acc ++ events, new_timeout, idle_timeout) after - min(remaining_timeout, 200) -> + min(remaining_timeout, idle_timeout) -> acc end end
Fix warnings over deprecated comment syntax
Fix documentation type in EventStore Usage Guide
…d SubscriptionState. Clarify behavior during max capacity and timer management for better understanding of event processing flow.
…nFsm. Introduce a dedicated function to cancel all buffer flush timers and enhance documentation to clarify timer behavior during event processing, ensuring events are flushed with bounded latency even when subscribers are at capacity.
…city The buffer_flush_after feature had three critical bugs causing event loss: 1. max_capacity state dropped new events from storage (no notify_events handler) 2. Subscription ignored max_capacity state, preventing fetch loop continuation 3. flush_buffer in max_capacity didn't attempt sending, preventing timeout-based delivery The subscription must continue fetching events even when subscriber is at capacity, queueing them until the subscriber ACKs. The timeout handler must attempt delivery and restart the timer if events remain, ensuring bounded latency even with back-pressure. Add comprehensive correctness tests to verify all events are delivered exactly once with proper ordering and no loss during back-pressure scenarios.
Add 23 additional tests covering: 1. No duplicates - Verify same event never sent twice across all scenarios 2. Latency bounds - Events delivered within timeout windows with/without back-pressure 3. Partition independence - Each partition maintains separate timer lifecycle 4. Edge cases - Single events, exact buffer matches, zero timeout, large buffers 5. Event ordering - Sequential delivery within partitions and with partitions 6. Rapid state transitions - Many quick append/ack cycles without event loss 7. Subscription lifecycle - Timers cancelled on unsubscribe, cleanup correctness 8. No event loss scenarios - Timeout cycles, max_capacity, concurrent appends 9. Integration - Works with checkpoint_after, selector filters, etc. Total test coverage now: 63 tests across 4 test suites - subscription_buffer_flush_after_test.exs (28 tests) - subscription_buffer_correctness_focus_test.exs (9 tests) - subscription_buffer_flush_diagnostics_test.exs (2 tests) - subscription_buffer_comprehensive_test.exs (23 tests) All tests pass with zero event loss or duplicates in all scenarios.
Document all 63 tests across 4 test suites with detailed breakdown of correctness properties verified, test scenarios, and implementation quality. Covers: delivery guarantees, latency bounds, state machine correctness, edge cases, and integration scenarios.
Add 36 additional tests covering: INVARIANT TESTS (19 tests): - Event number sequence integrity (no gaps across all scenarios) - Stream version sequencing - Batch composition (no event in multiple batches) - Batch size bounds verification - Event ordering across batches - Stress testing (100 events, 20 partitions, 30 rapid cycles) - Timing precision and latency bounds - Batch boundary properties - State consistency invariants - Recovery and cleanup correctness EDGE CASE TESTS (17 tests): - Exact boundary conditions (buffer_size == event_count) - Off-by-one scenarios - Configuration extremes (zero timeout, 10ms timeout, 5s timeout) - Very large buffers (1000) and very small buffers (1) - Interleaved operations (append during timeout, ack during fire) - Special stream patterns (single events per batch, alternating batches) - Concurrent timing scenarios (multiple timers firing) - Continuous stream processing - Large single appends (500 events) - Recovery from slow processing Total test coverage: 99 tests across 5 suites proving: ✅ Event delivery guarantees (no loss, no duplicates, ordering) ✅ Latency bounds under all conditions ✅ Partition independence ✅ State machine correctness ✅ Invariant preservation ✅ Edge case handling ✅ Stress test resilience
…ng 100% verification Add 5 comprehensive test suites covering: - Checkpoint & resume integration (7 tests) - Selector/filter completeness (14 tests) - Catch-up mode behavior (13 tests) - Subscription isolation & concurrency (7 tests) - Large scale stress testing (17 tests) Total: 121 tests, 100% passing, verifying: ✅ All events delivered exactly once ✅ Bounded latency maintained under all conditions ✅ Checkpoint safety without replays ✅ Selector filtering maintains all guarantees ✅ Catch-up mode transitions are safe ✅ Correctness at scale (50+ partitions, 500+ events) This achieves complete correctness verification across all delivery guarantees, edge cases, and advanced feature combinations.
The subscription FSM was crashing with FunctionClauseError when a buffer_flush_after timer fired while the subscription was in catching_up or request_catch_up states. Added handlers to clear the timer and remain in the current state, plus a catch-all handler for safety. Also includes: - Test timeout adjustments for CI reliability - Deterministic subscriber sorting (add pid as tiebreaker) - Fix test.all task to exclude slow tests in first run
- Remove duplicate catch-all flush_buffer handler (lines 405-414 was identical to lines 368-373) - Remove dead catch_up cast in max_capacity state - the catch_up event has no handler in max_capacity, so it fell through to a no-op. Events arrive via PubSub notify_events, not storage fetching. - Revert .tool-versions to original (elixir 1.16.0-otp-26, erlang 26.2.1)
- Removed unnecessary acknowledgment calls in checkpoint resume tests to streamline event processing. - Enhanced sequence integrity checks in edge cases tests to ensure events are in the expected order. - Clarified comments in large scale tests to better explain the expected event patterns and totals. - Updated assertions in selector completeness tests for more precise validation of event filtering.
8159e8d to
6799dd3
Compare
| fsm_state = subscription_struct.subscription | ||
| fsm_state.data | ||
| end | ||
| end |
There was a problem hiding this comment.
Manual diagnostic test committed to suite
Medium Severity
test/subscriptions/subscription_buffer_flush_diagnostics_test.exs introduces a diagnostic test module with @moduletag :manual plus multiple IO.inspect/2 and IO.puts/1 calls. If the test runner isn’t explicitly excluding the :manual tag, this can add noisy output and increase flakiness/CI runtime for normal test runs.
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
test/subscriptions/subscription_buffer_concurrent_subscribers_test.exs (1)
203-256:⚠️ Potential issue | 🟡 MinorAck ordering blocks further deliveries in the buffer_size=1 test.
In Lines 215‑251, ACKs are sent for prior batches, so the next batch may never arrive. ACK the current batch immediately and assert all three batches were received.
🛠️ Suggested fix
- receive do - {:events, b} -> b + receive do + {:events, b} -> + Subscription.ack(subscription, b) + b after 1000 -> [] end - receive do - {:events, b} -> - Subscription.ack(subscription, Enum.at(batches, 0)) - b + receive do + {:events, b} -> + Subscription.ack(subscription, b) + b after 1000 -> [] end - receive do - {:events, b} -> - Subscription.ack(subscription, Enum.at(batches, 1)) - b + receive do + {:events, b} -> + Subscription.ack(subscription, b) + b after 1000 -> [] end - receive do - {:events, _b} -> Subscription.ack(subscription, Enum.at(batches, 2)) + receive do + {:events, b} -> Subscription.ack(subscription, b) after 1000 -> nil end + assert length(batches) == 3🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/subscriptions/subscription_buffer_concurrent_subscribers_test.exs` around lines 203 - 256, The test's helper collect_and_ack_with_timeout delays acknowledging the current batch because it accumulates events and ACKs prior batches, which with buffer_size=1 blocks further deliveries; update the logic so that when receive {:events, events} from the subscription_pid you immediately call Subscription.ack(subscription_pid, events) for that batch (do not wait to ACK earlier batches first), add those events to the accumulator, and continue with the remaining timeout; update tests that call collect_and_ack_events/collect_and_ack_with_timeout to assert that all three batches were received (e.g., length or batch count) after the immediate ACK behavior.
♻️ Duplicate comments (5)
test/subscriptions/subscription_buffer_concurrent_subscribers_test.exs (1)
198-232:⚠️ Potential issue | 🟡 MinorTimeout helper returns after the first idle window.
Lines 229‑230 return
accimmediately, which can drop late events and make tests flaky. Consider recursing untilremaining_timeoutis exhausted.🛠️ Suggested fix
after min(remaining_timeout, 200) -> - acc + elapsed = System.monotonic_time(:millisecond) - start + new_timeout = remaining_timeout - elapsed + collect_and_ack_with_timeout(subscription_pid, acc, new_timeout) end🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/subscriptions/subscription_buffer_concurrent_subscribers_test.exs` around lines 198 - 232, The helper collect_and_ack_with_timeout can return early when the receive after fires (min(remaining_timeout, 200)) and drop late events; modify the after branch in collect_and_ack_with_timeout to not immediately return acc but instead compute sleep = min(remaining_timeout, 200) and, if remaining_timeout > sleep, recurse calling collect_and_ack_with_timeout(subscription_pid, acc, remaining_timeout - sleep), otherwise return acc—this ensures you loop until remaining_timeout is truly exhausted while still bounding idle wait to 200ms per iteration.test/subscriptions/subscription_buffer_invariants_test.exs (1)
390-417: State-consistency tests still don't validate the stated invariants.These tests only assert event counts (
length(events) > 0andlength(events) == 10), which is already covered extensively elsewhere. The moduledoc claims to verify "Last_received >= last_sent >= last_ack" and "checkpoint progress matches acked events," but neither test inspects subscription state. Consider using:sys.get_state(as done in the diagnostics test file,get_subscription_state/1) to assert on actual internal state, or remove these tests to avoid false confidence.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/subscriptions/subscription_buffer_invariants_test.exs` around lines 390 - 417, The tests subscribe_to_all_streams and collect_and_ack_events currently only assert event counts and do not validate the claimed invariants; update the two tests to fetch the subscription process state via get_subscription_state(subscription) (or :sys.get_state/1) and assert the internal fields last_received, last_sent, last_ack and checkpoint_progress satisfy last_received >= last_sent and last_sent >= last_ack, and that checkpoint_progress matches the number of acked events, or if you prefer remove these misleading tests entirely; locate the tests referring to subscribe_to_all_streams, collect_and_ack_events, and get_subscription_state to implement the state assertions or deletion.test/subscriptions/subscription_buffer_selector_completeness_test.exs (1)
371-389:collect_and_ack_with_timeoutearly-exit still not addressed.The
afterclause returnsaccimmediately on a 200ms idle window rather than continuing to loop untilremaining_timeoutis fully consumed. This was flagged in a prior review.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/subscriptions/subscription_buffer_selector_completeness_test.exs` around lines 371 - 389, The after clause in collect_and_ack_with_timeout currently returns acc after a short min(remaining_timeout, 200) idle window, which exits early; change the after branch to compute elapsed time (using System.monotonic_time(:millisecond) - start), subtract it from remaining_timeout to produce new_timeout, and recursively call collect_and_ack_with_timeout(subscription_pid, acc, new_timeout) so the function continues looping until the remaining_timeout <= 0 matching the existing guard; keep the {:events, events} handling and Subscription.ack(subscription_pid, events) as-is.test/subscriptions/subscription_buffer_checkpoint_resume_test.exs (2)
288-290:⚠️ Potential issue | 🟡 MinorTighten the “only new events” assertion.
Enum.all?(nums, &(&1 > 2))can still pass if earlier events replay; it’s too loose. Assert the exact expected range to avoid false positives.✅ More precise assertion
- nums = Enum.map(events2, & &1.event_number) - assert Enum.all?(nums, &(&1 > 2)), "Should only receive new events after checkpoint" + nums = + events2 + |> Enum.map(& &1.event_number) + |> Enum.sort() + + assert nums == Enum.to_list(7..12), + "Should only receive new events after checkpoint"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/subscriptions/subscription_buffer_checkpoint_resume_test.exs` around lines 288 - 290, The assertion checking for "only new events" is too loose; replace the fuzzy check on nums (from events2) with an exact expectation of the event_number sequence after the checkpoint (e.g., assert nums equals the specific list or range of expected numbers 3..8) so the test fails if any earlier events are replayed; locate the variable nums produced from events2.event_number and assert it exactly matches the expected list or Enum.to_list(3..8).
210-231:⚠️ Potential issue | 🟡 MinorAvoid hard-coded flush timing; compare to configured checkpoint threshold.
The
elapsed < 250assertion is brittle and can flake on slower CI. Prefer using the configuredcheckpoint_after(or derived slack) so the test reflects the actual configuration.💡 Suggested adjustment
- {:ok, subscription} = + checkpoint_after = 500 + buffer_flush_after = 100 + {:ok, subscription} = EventStore.subscribe_to_all_streams( subscription_name, self(), buffer_size: 10, - buffer_flush_after: 100, - checkpoint_after: 500 + buffer_flush_after: buffer_flush_after, + checkpoint_after: checkpoint_after ) @@ - assert elapsed < 250, "Should flush via timeout, not wait for checkpoint" + assert elapsed < checkpoint_after, + "Should flush via timeout, not wait for checkpoint"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/subscriptions/subscription_buffer_checkpoint_resume_test.exs` around lines 210 - 231, The test uses a brittle hard-coded assertion (assert elapsed < 250) to verify flush-by-time behavior; update it to compare against the configured checkpoint_after value used when calling EventStore.subscribe_to_all_streams (checkpoint_after: 500) or a derived slack (e.g., checkpoint_after / 2 or checkpoint_after - 100) so the assertion reflects the actual subscription configuration; locate the subscription creation (EventStore.subscribe_to_all_streams) and replace the literal 250 check with a comparison to the checkpoint_after-derived threshold.
🧹 Nitpick comments (3)
test/subscriptions/subscription_buffer_flush_diagnostics_test.exs (1)
23-23: Diagnostic tests contain extensiveIO.inspect/IO.putsdebug output.While the
@moduletag :manualprevents these from running in CI, the heavy use ofIO.inspectandIO.puts(lines 23, 30, 47, 66, 71–72, 80–81, 103–104, 107) is atypical for test modules. Consider usingLogger.debugor removing the IO calls and relying on assertions + ExUnit's built-in failure output instead.Also applies to: 30-30, 47-47, 66-66, 71-72, 80-81, 103-104, 107-107
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/subscriptions/subscription_buffer_flush_diagnostics_test.exs` at line 23, Replace the ad-hoc IO.inspect/IO.puts debug output in subscription_buffer_flush_diagnostics_test.exs with proper logging or remove them: find the IO.inspect/IO.puts calls (e.g., IO.inspect(batch1, label: "Batch 1")) and either delete them or change them to Logger.debug/Logger.info calls and add a require Logger at the top of the test module; ensure assertions remain sufficient so test behavior is unchanged and run the tests to confirm no extraneous output remains.guides/BufferFlushArchitecture.md (1)
74-104: Add language identifiers to fenced code blocks.Static analysis flags several fenced code blocks (lines 74, 155, 171, 263, 333, 353) without a language specified. For ASCII diagrams and execution traces, use
```textto satisfy the linter.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@guides/BufferFlushArchitecture.md` around lines 74 - 104, The fenced code blocks in BufferFlushArchitecture.md (including the ASCII diagram shown in the diff and other blocks referenced at lines 74, 155, 171, 263, 333, 353) are missing language identifiers; update each triple-backtick fence for ASCII diagrams and execution traces to use a text language tag (i.e., change ``` to ```text) so the linter accepts them, making sure to update the blocks containing the Subscription FSM diagram and any execution trace blocks consistently.test/subscriptions/subscription_buffer_correctness_focus_test.exs (1)
281-316: Extract shared test helpers into a common module to reduce duplication.The helpers
subscribe_to_all_streams/1,append_to_stream/3,collect_and_ack_events/2, andcollect_and_ack_with_timeout/3are copy-pasted verbatim across at least 10 new test files in this PR (e.g.,subscription_buffer_large_scale_test.exs,subscription_buffer_comprehensive_test.exs,subscription_buffer_edge_cases_test.exs,subscription_buffer_invariants_test.exs, etc.). Consider extracting them into a shared helper module (e.g.,EventStore.Subscriptions.TestHelpers) andimport-ing it, so fixes (like thecollect_and_ack_with_timeoutbehavior noted below) only need to be applied once.Additionally,
collect_and_ack_with_timeout/3here (and in most other files) exits immediately after a 200ms idle window, even ifremaining_timeouthasn't elapsed. The version insubscription_buffer_invariants_test.exs(lines 485–500) correctly continues the loop. This inconsistency could cause flaky tests on slow CI where inter-batch gaps exceed 200ms.Example shared helper module
# test/support/subscription_test_helpers.ex defmodule EventStore.Subscriptions.TestHelpers do alias EventStore.{EventFactory, UUID} alias EventStore.Subscriptions.Subscription alias TestEventStore, as: EventStore import ExUnit.Assertions def subscribe_to_all_streams(opts) do subscription_name = UUID.uuid4() {:ok, subscription} = EventStore.subscribe_to_all_streams(subscription_name, self(), opts) assert_receive {:subscribed, ^subscription} {:ok, subscription} end def append_to_stream(stream_uuid, event_count, expected_version \\ 0) do events = EventFactory.create_events(event_count, expected_version + 1) :ok = EventStore.append_to_stream(stream_uuid, expected_version, events) end def collect_and_ack_events(subscription_pid, timeout: timeout) do collect_and_ack_with_timeout(subscription_pid, [], timeout) end defp collect_and_ack_with_timeout(_pid, acc, remaining) when remaining <= 0, do: acc defp collect_and_ack_with_timeout(pid, acc, remaining) do start = System.monotonic_time(:millisecond) receive do {:events, events} -> :ok = Subscription.ack(pid, events) elapsed = System.monotonic_time(:millisecond) - start collect_and_ack_with_timeout(pid, acc ++ events, remaining - elapsed) after min(remaining, 200) -> elapsed = System.monotonic_time(:millisecond) - start collect_and_ack_with_timeout(pid, acc, remaining - elapsed) end end end🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@test/subscriptions/subscription_buffer_correctness_focus_test.exs` around lines 281 - 316, Extract the duplicated helpers subscribe_to_all_streams/1, append_to_stream/3, collect_and_ack_events/2 and collect_and_ack_with_timeout/3 into a shared test helper module (e.g., EventStore.Subscriptions.TestHelpers) and import that module in each test file so fixes are centralized; while extracting, fix collect_and_ack_with_timeout/3 so the after clause does not return immediately — subtract the elapsed time and loop back into collect_and_ack_with_timeout/3 (instead of returning acc) so the remaining_timeout is respected even when the receive times out for up to 200ms; keep references to Subscription.ack/2, EventFactory.create_events/2, EventStore.subscribe_to_all_streams/3 and EventStore.append_to_stream/3 when moving logic.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@guides/Subscriptions.md`:
- Around line 355-529: Three fenced code blocks (the stream list block under
"The Flow", the per-partition timer timeline, and the checkpoint advances list)
are missing language tags; add a language specifier such as `text` to each of
those triple-backtick fences (the blocks showing "Stream-A: Event `#100`..." ,
"T=0s: All events buffered..." , and "After ACK 102: Checkpoint cannot
advance...") so markdownlint MD040 is satisfied.
In `@test/subscriptions/subscription_buffer_catchup_mode_test.exs`:
- Around line 60-78: The test uses a hard-coded latency bound (assert elapsed <
300) which is flaky; compute the expected max from the configured
buffer_flush_after used in subscribe_to_all_streams (buffer_flush_after: 100)
and assert elapsed is less than buffer_flush_after plus a small slack (e.g.
+100ms) instead of 300ms. Update the "catch-up respects buffer_flush_after
timeout" test to derive the threshold from the buffer_flush_after variable and
apply the same pattern to the other timing assertions referenced around the
366-405 region in this file.
In `@test/subscriptions/subscription_buffer_edge_cases_test.exs`:
- Around line 386-421: The helper collect_and_ack_with_timeout currently returns
acc when the receive times out (after min(remaining_timeout, 200)), which can
drop late events; change the after branch so it does not return immediately but
subtracts the actual wait interval from remaining_timeout and calls
collect_and_ack_with_timeout(subscription_pid, acc, new_remaining_timeout) until
remaining_timeout <= 0. Update the after clause in collect_and_ack_with_timeout
(and ensure collect_and_ack_events still calls it) to compute the wait duration
(min(remaining_timeout, 200)), reduce remaining_timeout by that amount, and
continue looping rather than returning on first idle window.
In `@test/subscriptions/subscription_buffer_flush_diagnostics_test.exs`:
- Around line 32-40: The test currently uses IO.puts to signal an unexpected
message in the receive block (the {:events, batch2} clause), which doesn't fail
the test; change this to a proper assertion by either replacing the whole
receive/do...after with Elixir's refute_receive {:events, _batch2}, 200 or, if
you keep the receive, call flunk("Unexpected events received while at capacity:
#{inspect(batch2)}") inside the {:events, batch2} clause; reference the
{:events, batch2} pattern and use refute_receive/2 or flunk/1 to ensure the test
actually fails on unexpected events.
- Around line 111-128: The helper collect_with_logging mismanages the timeout
and never acknowledges events: change it to compute elapsed time using monotonic
timestamps (capture start before receive and subtract actual delta from
remaining_timeout) instead of subtracting a fixed 100, and after receiving
events send the appropriate acknowledgement back to the subscription (use the
subscription_pid passed into collect_with_logging to ack the batch or each event
so the subscriber buffer can free up) before recursing; keep the same function
names (collect_with_logging, subscription_pid, remaining_timeout) so the fix is
easy to locate.
---
Outside diff comments:
In `@test/subscriptions/subscription_buffer_concurrent_subscribers_test.exs`:
- Around line 203-256: The test's helper collect_and_ack_with_timeout delays
acknowledging the current batch because it accumulates events and ACKs prior
batches, which with buffer_size=1 blocks further deliveries; update the logic so
that when receive {:events, events} from the subscription_pid you immediately
call Subscription.ack(subscription_pid, events) for that batch (do not wait to
ACK earlier batches first), add those events to the accumulator, and continue
with the remaining timeout; update tests that call
collect_and_ack_events/collect_and_ack_with_timeout to assert that all three
batches were received (e.g., length or batch count) after the immediate ACK
behavior.
---
Duplicate comments:
In `@test/subscriptions/subscription_buffer_checkpoint_resume_test.exs`:
- Around line 288-290: The assertion checking for "only new events" is too
loose; replace the fuzzy check on nums (from events2) with an exact expectation
of the event_number sequence after the checkpoint (e.g., assert nums equals the
specific list or range of expected numbers 3..8) so the test fails if any
earlier events are replayed; locate the variable nums produced from
events2.event_number and assert it exactly matches the expected list or
Enum.to_list(3..8).
- Around line 210-231: The test uses a brittle hard-coded assertion (assert
elapsed < 250) to verify flush-by-time behavior; update it to compare against
the configured checkpoint_after value used when calling
EventStore.subscribe_to_all_streams (checkpoint_after: 500) or a derived slack
(e.g., checkpoint_after / 2 or checkpoint_after - 100) so the assertion reflects
the actual subscription configuration; locate the subscription creation
(EventStore.subscribe_to_all_streams) and replace the literal 250 check with a
comparison to the checkpoint_after-derived threshold.
In `@test/subscriptions/subscription_buffer_concurrent_subscribers_test.exs`:
- Around line 198-232: The helper collect_and_ack_with_timeout can return early
when the receive after fires (min(remaining_timeout, 200)) and drop late events;
modify the after branch in collect_and_ack_with_timeout to not immediately
return acc but instead compute sleep = min(remaining_timeout, 200) and, if
remaining_timeout > sleep, recurse calling
collect_and_ack_with_timeout(subscription_pid, acc, remaining_timeout - sleep),
otherwise return acc—this ensures you loop until remaining_timeout is truly
exhausted while still bounding idle wait to 200ms per iteration.
In `@test/subscriptions/subscription_buffer_invariants_test.exs`:
- Around line 390-417: The tests subscribe_to_all_streams and
collect_and_ack_events currently only assert event counts and do not validate
the claimed invariants; update the two tests to fetch the subscription process
state via get_subscription_state(subscription) (or :sys.get_state/1) and assert
the internal fields last_received, last_sent, last_ack and checkpoint_progress
satisfy last_received >= last_sent and last_sent >= last_ack, and that
checkpoint_progress matches the number of acked events, or if you prefer remove
these misleading tests entirely; locate the tests referring to
subscribe_to_all_streams, collect_and_ack_events, and get_subscription_state to
implement the state assertions or deletion.
In `@test/subscriptions/subscription_buffer_selector_completeness_test.exs`:
- Around line 371-389: The after clause in collect_and_ack_with_timeout
currently returns acc after a short min(remaining_timeout, 200) idle window,
which exits early; change the after branch to compute elapsed time (using
System.monotonic_time(:millisecond) - start), subtract it from remaining_timeout
to produce new_timeout, and recursively call
collect_and_ack_with_timeout(subscription_pid, acc, new_timeout) so the function
continues looping until the remaining_timeout <= 0 matching the existing guard;
keep the {:events, events} handling and Subscription.ack(subscription_pid,
events) as-is.
---
Nitpick comments:
In `@guides/BufferFlushArchitecture.md`:
- Around line 74-104: The fenced code blocks in BufferFlushArchitecture.md
(including the ASCII diagram shown in the diff and other blocks referenced at
lines 74, 155, 171, 263, 333, 353) are missing language identifiers; update each
triple-backtick fence for ASCII diagrams and execution traces to use a text
language tag (i.e., change ``` to ```text) so the linter accepts them, making
sure to update the blocks containing the Subscription FSM diagram and any
execution trace blocks consistently.
In `@test/subscriptions/subscription_buffer_correctness_focus_test.exs`:
- Around line 281-316: Extract the duplicated helpers
subscribe_to_all_streams/1, append_to_stream/3, collect_and_ack_events/2 and
collect_and_ack_with_timeout/3 into a shared test helper module (e.g.,
EventStore.Subscriptions.TestHelpers) and import that module in each test file
so fixes are centralized; while extracting, fix collect_and_ack_with_timeout/3
so the after clause does not return immediately — subtract the elapsed time and
loop back into collect_and_ack_with_timeout/3 (instead of returning acc) so the
remaining_timeout is respected even when the receive times out for up to 200ms;
keep references to Subscription.ack/2, EventFactory.create_events/2,
EventStore.subscribe_to_all_streams/3 and EventStore.append_to_stream/3 when
moving logic.
In `@test/subscriptions/subscription_buffer_flush_diagnostics_test.exs`:
- Line 23: Replace the ad-hoc IO.inspect/IO.puts debug output in
subscription_buffer_flush_diagnostics_test.exs with proper logging or remove
them: find the IO.inspect/IO.puts calls (e.g., IO.inspect(batch1, label: "Batch
1")) and either delete them or change them to Logger.debug/Logger.info calls and
add a require Logger at the top of the test module; ensure assertions remain
sufficient so test behavior is unchanged and run the tests to confirm no
extraneous output remains.
ℹ️ Review info
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (23)
.tool-versionsguides/BufferFlushArchitecture.mdguides/Subscriptions.mdlib/event_store.exlib/event_store/storage/snapshot.exlib/event_store/subscriptions/subscription.exlib/event_store/subscriptions/subscription_fsm.exlib/event_store/subscriptions/subscription_state.extest/shared_connection_pool_test.exstest/storage/append_events_test.exstest/storage/stream_persistence_test.exstest/subscriptions/concurrent_subscription_test.exstest/subscriptions/subscription_buffer_catchup_mode_test.exstest/subscriptions/subscription_buffer_checkpoint_resume_test.exstest/subscriptions/subscription_buffer_comprehensive_test.exstest/subscriptions/subscription_buffer_concurrent_subscribers_test.exstest/subscriptions/subscription_buffer_correctness_focus_test.exstest/subscriptions/subscription_buffer_edge_cases_test.exstest/subscriptions/subscription_buffer_flush_after_test.exstest/subscriptions/subscription_buffer_flush_diagnostics_test.exstest/subscriptions/subscription_buffer_invariants_test.exstest/subscriptions/subscription_buffer_large_scale_test.exstest/subscriptions/subscription_buffer_selector_completeness_test.exs
🚧 Files skipped from review as they are similar to previous changes (5)
- test/storage/append_events_test.exs
- lib/event_store/storage/snapshot.ex
- .tool-versions
- lib/event_store.ex
- lib/event_store/subscriptions/subscription_state.ex
| ## Buffer Flush Behavior | ||
|
|
||
| The `buffer_flush_after` option provides bounded latency guarantees for event delivery by automatically flushing buffered events after a timeout period. This is particularly useful when using `buffer_size > 1` for throughput optimization but still requiring predictable latency during low-traffic periods. | ||
|
|
||
| ### How It Works | ||
|
|
||
| #### Without `buffer_flush_after` | ||
|
|
||
| ```elixir | ||
| {:ok, subscription} = | ||
| EventStore.subscribe_to_all_streams("my_sub", self(), | ||
| buffer_size: 100 | ||
| ) | ||
| ``` | ||
|
|
||
| **Behavior:** | ||
| - Events are buffered until 100 events accumulate | ||
| - During high traffic: ✅ Batches flush quickly (good throughput) | ||
| - During low traffic: ❌ Events wait indefinitely for the 100th event | ||
| - **Problem:** Read models can become stale during quiet periods | ||
|
|
||
| #### With `buffer_flush_after` | ||
|
|
||
| ```elixir | ||
| {:ok, subscription} = | ||
| EventStore.subscribe_to_all_streams("my_sub", self(), | ||
| buffer_size: 100, | ||
| buffer_flush_after: 5_000 # 5 seconds | ||
| ) | ||
| ``` | ||
|
|
||
| **Behavior:** | ||
| - Events are buffered until 100 events **OR** 5 seconds, whichever comes first | ||
| - During high traffic: ✅ Batches flush when full (good throughput) | ||
| - During low traffic: ✅ Partial batches flush after 5s (bounded latency) | ||
| - **Result:** Predictable latency regardless of traffic patterns | ||
|
|
||
| ### Per-Partition Timers | ||
|
|
||
| When using `partition_by`, each partition maintains its own independent timer: | ||
|
|
||
| ```elixir | ||
| {:ok, subscription} = | ||
| EventStore.subscribe_to_all_streams("my_sub", self(), | ||
| buffer_size: 100, | ||
| buffer_flush_after: 5_000, | ||
| partition_by: fn event -> event.stream_uuid end | ||
| ) | ||
| ``` | ||
|
|
||
| **Why per-partition timers are necessary:** | ||
|
|
||
| Consider a scenario with different traffic patterns per partition: | ||
| - Stream A: 1000 events/second (high volume) | ||
| - Stream B: 1 event/minute (low volume) | ||
|
|
||
| **With per-partition timers (current design):** | ||
| - Stream A: Buffer fills quickly → flushes on `buffer_size` | ||
| - Stream B: Buffer doesn't fill → timer fires after 5s → flushes partial batch | ||
| - ✅ Both streams get timely delivery | ||
|
|
||
| **Without per-partition timers (hypothetical):** | ||
| - Stream A: Buffer fills quickly → flushes → **resets global timer** | ||
| - Stream B: Waits for global timer → **but Stream A keeps resetting it!** | ||
| - ❌ Stream B's events never time out → stale data | ||
|
|
||
| ### Acknowledgement and Checkpointing | ||
|
|
||
| **Important:** While partitions have independent flush timers, acknowledgements and checkpoints respect **global event ordering**. | ||
|
|
||
| #### The Flow | ||
|
|
||
| 1. **Events arrive** across multiple partitions: | ||
| ``` | ||
| Stream-A: Event #100, #101, #104 | ||
| Stream-B: Event #102 | ||
| Stream-C: Event #103 | ||
| ``` | ||
|
|
||
| 2. **Per-partition timers** control when events are sent to subscribers: | ||
| ``` | ||
| T=0s: All events buffered | ||
| T=5s: Stream-B timer fires → Event #102 sent to subscriber | ||
| T=5.1s: Stream-A timer fires → Events #100, #101, #104 sent | ||
| T=10s: Stream-C timer fires → Event #103 sent | ||
| ``` | ||
|
|
||
| 3. **Subscriber acknowledges** events: | ||
| ```elixir | ||
| # Subscriber receives Stream-B events first (due to timer) | ||
| {:events, [event_102]} -> :ok = EventStore.ack(subscription, event_102) | ||
|
|
||
| # Then Stream-A events | ||
| {:events, [event_100, event_101, event_104]} -> | ||
| :ok = EventStore.ack(subscription, event_104) # ACKs all in batch | ||
|
|
||
| # Finally Stream-C events | ||
| {:events, [event_103]} -> :ok = EventStore.ack(subscription, event_103) | ||
| ``` | ||
|
|
||
| 4. **Checkpoint advances** in global event order: | ||
| ``` | ||
| After ACK 102: Checkpoint cannot advance (event 100 not ACK'd yet) | ||
| After ACK 104: Checkpoint advances to 102 (100, 101, 102 all ACK'd) | ||
| After ACK 103: Checkpoint advances to 104 (all events ACK'd) | ||
| ``` | ||
|
|
||
| **Key insight:** Events from different partitions can be **delivered at different times**, but the checkpoint always advances in **global event number order** to ensure consistent replay on restart. | ||
|
|
||
| ### Use Cases | ||
|
|
||
| #### Read Model Projections with Batching | ||
|
|
||
| ```elixir | ||
| defmodule MyApp.ReadModelProjector do | ||
| use Commanded.Event.Handler, | ||
| application: MyApp, | ||
| name: __MODULE__, | ||
| batch_size: 1000, # Batch for database performance | ||
| buffer_flush_after: 5_000 # But don't wait forever | ||
|
|
||
| def handle_batch(events) do | ||
| Repo.transaction(fn -> | ||
| # Insert 1000 events efficiently | ||
| Enum.each(events, &insert_into_read_model/1) | ||
| end) | ||
| :ok | ||
| end | ||
| end | ||
| ``` | ||
|
|
||
| **Benefits:** | ||
| - High traffic: Efficient 1000-event batches | ||
| - Low traffic: Events still delivered within 5 seconds | ||
| - Predictable read model freshness | ||
|
|
||
| #### Per-Stream Processing with Variable Traffic | ||
|
|
||
| ```elixir | ||
| {:ok, subscription} = | ||
| EventStore.subscribe_to_all_streams("processor", self(), | ||
| buffer_size: 50, | ||
| buffer_flush_after: 3_000, | ||
| partition_by: fn event -> event.stream_uuid end, | ||
| concurrency_limit: 10 | ||
| ) | ||
| ``` | ||
|
|
||
| **Benefits:** | ||
| - Each stream processed independently | ||
| - High-volume streams don't block low-volume streams | ||
| - All streams get 3-second latency guarantee | ||
|
|
||
| ### Configuration Guidelines | ||
|
|
||
| **Choose `buffer_size` based on throughput needs:** | ||
| - `buffer_size: 1` (default) - Lowest latency, no batching needed | ||
| - `buffer_size: 10-100` - Good balance for most use cases | ||
| - `buffer_size: 1000+` - High-throughput batch processing | ||
|
|
||
| **Choose `buffer_flush_after` based on latency requirements:** | ||
| - `buffer_flush_after: 0` (default) - No timeout (only flush on buffer_size) | ||
| - `buffer_flush_after: 1_000` - 1 second max latency (real-time systems) | ||
| - `buffer_flush_after: 5_000` - 5 second max latency (typical read models) | ||
| - `buffer_flush_after: 30_000` - 30 second max latency (background processing) | ||
|
|
||
| **Rule of thumb:** | ||
| ```elixir | ||
| # If you set buffer_size > 1, you probably want buffer_flush_after too | ||
| {:ok, subscription} = | ||
| EventStore.subscribe_to_all_streams("my_sub", self(), | ||
| buffer_size: 100, | ||
| buffer_flush_after: 5_000 # Don't let events sit indefinitely! | ||
| ) | ||
| ``` |
There was a problem hiding this comment.
Add language tags to fenced blocks (MD040).
Lines 428, 435, and 456 contain fenced blocks without a language specifier; markdownlint flags these. Please add a language such as text for non-code diagrams.
🔧 Suggested fix
- ```
+ ```text
Stream-A: Event `#100`, `#101`, `#104`
Stream-B: Event `#102`
Stream-C: Event `#103`T=0s: All events buffered
T=5s: Stream-B timer fires → Event #102 sent to subscriber
T=5.1s: Stream-A timer fires → Events #100, #101, #104 sent
T=10s: Stream-C timer fires → Event #103 sent
- ```
+ ```text
After ACK 102: Checkpoint cannot advance (event 100 not ACK'd yet)
After ACK 104: Checkpoint advances to 102 (100, 101, 102 all ACK'd)
After ACK 103: Checkpoint advances to 104 (all events ACK'd)
</details>
<details>
<summary>🧰 Tools</summary>
<details>
<summary>🪛 markdownlint-cli2 (0.21.0)</summary>
[warning] 428-428: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
---
[warning] 435-435: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
---
[warning] 456-456: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
</details>
</details>
<details>
<summary>🤖 Prompt for AI Agents</summary>
Verify each finding against the current code and only fix it if needed.
In @guides/Subscriptions.md around lines 355 - 529, Three fenced code blocks
(the stream list block under "The Flow", the per-partition timer timeline, and
the checkpoint advances list) are missing language tags; add a language
specifier such as text to each of those triple-backtick fences (the blocks
showing "Stream-A: Event #100..." , "T=0s: All events buffered..." , and "After
ACK 102: Checkpoint cannot advance...") so markdownlint MD040 is satisfied.
</details>
<!-- fingerprinting:phantom:medusa:eagle -->
<!-- This is an auto-generated comment by CodeRabbit -->
| test "catch-up respects buffer_flush_after timeout" do | ||
| {:ok, subscription} = | ||
| subscribe_to_all_streams( | ||
| buffer_size: 10, | ||
| buffer_flush_after: 100 | ||
| ) | ||
|
|
||
| # Append fewer than buffer_size | ||
| append_to_stream("stream1", 3) | ||
|
|
||
| # Should still flush via timeout during catch-up or immediately if subscriber ready | ||
| start = System.monotonic_time(:millisecond) | ||
| assert_receive {:events, events}, 500 | ||
| elapsed = System.monotonic_time(:millisecond) - start | ||
|
|
||
| assert length(events) == 3 | ||
| # Should arrive within reasonable time (either via timeout or immediate delivery) | ||
| assert elapsed < 300, "Should deliver within reasonable latency" | ||
|
|
There was a problem hiding this comment.
Derive latency bounds from buffer_flush_after to reduce CI flakiness.
Hard-coded 300ms/250ms thresholds can intermittently fail under load. Use the configured buffer_flush_after plus a small slack to keep the assertions meaningful and robust.
💡 Example adjustment (apply similarly to other timing asserts)
- {:ok, subscription} =
- subscribe_to_all_streams(
- buffer_size: 10,
- buffer_flush_after: 100
- )
+ buffer_flush_after = 100
+ max_latency = buffer_flush_after + 200
+ {:ok, subscription} =
+ subscribe_to_all_streams(
+ buffer_size: 10,
+ buffer_flush_after: buffer_flush_after
+ )
@@
- assert elapsed < 300, "Should deliver within reasonable latency"
+ assert elapsed < max_latency, "Should deliver within reasonable latency"- {:ok, subscription} =
- subscribe_to_all_streams(
- buffer_size: 3,
- buffer_flush_after: 100
- )
+ buffer_flush_after = 100
+ max_latency = buffer_flush_after + 150
+ {:ok, subscription} =
+ subscribe_to_all_streams(
+ buffer_size: 3,
+ buffer_flush_after: buffer_flush_after
+ )
@@
- assert Enum.all?(timings, &(&1 < 250)),
+ assert Enum.all?(timings, &(&1 < max_latency)),
"Each delivery should be within latency bounds: #{inspect(timings)}"Also applies to: 366-405
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@test/subscriptions/subscription_buffer_catchup_mode_test.exs` around lines 60
- 78, The test uses a hard-coded latency bound (assert elapsed < 300) which is
flaky; compute the expected max from the configured buffer_flush_after used in
subscribe_to_all_streams (buffer_flush_after: 100) and assert elapsed is less
than buffer_flush_after plus a small slack (e.g. +100ms) instead of 300ms.
Update the "catch-up respects buffer_flush_after timeout" test to derive the
threshold from the buffer_flush_after variable and apply the same pattern to the
other timing assertions referenced around the 366-405 region in this file.
| # Helpers | ||
|
|
||
| defp subscribe_to_all_streams(opts) do | ||
| subscription_name = UUID.uuid4() | ||
| {:ok, subscription} = EventStore.subscribe_to_all_streams(subscription_name, self(), opts) | ||
| assert_receive {:subscribed, ^subscription} | ||
| {:ok, subscription} | ||
| end | ||
|
|
||
| defp append_to_stream(stream_uuid, event_count, expected_version \\ 0) do | ||
| events = EventFactory.create_events(event_count, expected_version + 1) | ||
| :ok = EventStore.append_to_stream(stream_uuid, expected_version, events) | ||
| end | ||
|
|
||
| defp collect_and_ack_events(subscription_pid, timeout: timeout) do | ||
| collect_and_ack_with_timeout(subscription_pid, [], timeout) | ||
| end | ||
|
|
||
| defp collect_and_ack_with_timeout(_subscription_pid, acc, remaining_timeout) | ||
| when remaining_timeout <= 0 do | ||
| acc | ||
| end | ||
|
|
||
| defp collect_and_ack_with_timeout(subscription_pid, acc, remaining_timeout) do | ||
| start = System.monotonic_time(:millisecond) | ||
|
|
||
| receive do | ||
| {:events, events} -> | ||
| :ok = Subscription.ack(subscription_pid, events) | ||
| elapsed = System.monotonic_time(:millisecond) - start | ||
| new_timeout = remaining_timeout - elapsed | ||
| collect_and_ack_with_timeout(subscription_pid, acc ++ events, new_timeout) | ||
| after | ||
| min(remaining_timeout, 200) -> | ||
| acc | ||
| end |
There was a problem hiding this comment.
Timeout helper returns on first idle window.
Lines 419‑420 return early, which can drop late events and make tests flaky. Consider continuing until remaining_timeout is exhausted.
🛠️ Suggested fix
after
min(remaining_timeout, 200) ->
- acc
+ elapsed = System.monotonic_time(:millisecond) - start
+ new_timeout = remaining_timeout - elapsed
+ collect_and_ack_with_timeout(subscription_pid, acc, new_timeout)
end🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@test/subscriptions/subscription_buffer_edge_cases_test.exs` around lines 386
- 421, The helper collect_and_ack_with_timeout currently returns acc when the
receive times out (after min(remaining_timeout, 200)), which can drop late
events; change the after branch so it does not return immediately but subtracts
the actual wait interval from remaining_timeout and calls
collect_and_ack_with_timeout(subscription_pid, acc, new_remaining_timeout) until
remaining_timeout <= 0. Update the after clause in collect_and_ack_with_timeout
(and ensure collect_and_ack_events still calls it) to compute the wait duration
(min(remaining_timeout, 200)), reduce remaining_timeout by that amount, and
continue looping rather than returning on first idle window.
| # Check if more events arrived | ||
| receive do | ||
| {:events, batch2} -> | ||
| IO.inspect(batch2, label: "Batch 2 (received while at capacity)") | ||
| IO.puts("ERROR: Should not have received events while at capacity!") | ||
| after | ||
| 200 -> | ||
| IO.puts("OK: No events received while at capacity (as expected)") | ||
| end |
There was a problem hiding this comment.
IO-based error message is not an actual assertion.
Line 36 prints "ERROR: Should not have received events while at capacity!" but doesn't fail the test. If events arrive unexpectedly, this test silently passes. Use flunk/1 or refute_receive instead.
Proposed fix
- receive do
- {:events, batch2} ->
- IO.inspect(batch2, label: "Batch 2 (received while at capacity)")
- IO.puts("ERROR: Should not have received events while at capacity!")
- after
- 200 ->
- IO.puts("OK: No events received while at capacity (as expected)")
- end
+ refute_receive {:events, _}, 200, "Should not receive events while at capacity"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@test/subscriptions/subscription_buffer_flush_diagnostics_test.exs` around
lines 32 - 40, The test currently uses IO.puts to signal an unexpected message
in the receive block (the {:events, batch2} clause), which doesn't fail the
test; change this to a proper assertion by either replacing the whole
receive/do...after with Elixir's refute_receive {:events, _batch2}, 200 or, if
you keep the receive, call flunk("Unexpected events received while at capacity:
#{inspect(batch2)}") inside the {:events, batch2} clause; reference the
{:events, batch2} pattern and use refute_receive/2 or flunk/1 to ensure the test
actually fails on unexpected events.
| defp collect_with_logging(_subscription_pid, acc, remaining_timeout: remaining) | ||
| when remaining <= 0 do | ||
| IO.puts("Timeout expired, stopping collection") | ||
| acc | ||
| end | ||
|
|
||
| defp collect_with_logging(subscription_pid, acc, remaining_timeout: remaining) do | ||
| receive do | ||
| {:events, events} -> | ||
| IO.puts("Received #{length(events)} events") | ||
| Enum.each(events, &IO.inspect(&1.event_number, label: " event_number")) | ||
| collect_with_logging(subscription_pid, acc ++ events, remaining_timeout: remaining - 100) | ||
| after | ||
| 200 -> | ||
| IO.puts("No events received in 200ms") | ||
| collect_with_logging(subscription_pid, acc, remaining_timeout: remaining - 200) | ||
| end | ||
| end |
There was a problem hiding this comment.
collect_with_logging has inaccurate timeout tracking and missing acks.
Two issues:
- Line 122 subtracts a fixed
100from the remaining timeout instead of measuring actual elapsed time, making the timeout budget inaccurate. - Events are never acknowledged, so with
buffer_size < event_count, the subscriber hits capacity and collection stalls after the first batch.
Since these are manual diagnostic tests, the impact is low, but fixing these would make the diagnostics more reliable.
Proposed fix
defp collect_with_logging(subscription_pid, acc, remaining_timeout: remaining) do
+ start = System.monotonic_time(:millisecond)
+
receive do
{:events, events} ->
IO.puts("Received #{length(events)} events")
Enum.each(events, &IO.inspect(&1.event_number, label: " event_number"))
- collect_with_logging(subscription_pid, acc ++ events, remaining_timeout: remaining - 100)
+ :ok = Subscription.ack(subscription_pid, events)
+ elapsed = System.monotonic_time(:millisecond) - start
+ collect_with_logging(subscription_pid, acc ++ events, remaining_timeout: remaining - elapsed)
after
200 ->
IO.puts("No events received in 200ms")
- collect_with_logging(subscription_pid, acc, remaining_timeout: remaining - 200)
+ elapsed = System.monotonic_time(:millisecond) - start
+ collect_with_logging(subscription_pid, acc, remaining_timeout: remaining - elapsed)
end
end🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@test/subscriptions/subscription_buffer_flush_diagnostics_test.exs` around
lines 111 - 128, The helper collect_with_logging mismanages the timeout and
never acknowledges events: change it to compute elapsed time using monotonic
timestamps (capture start before receive and subtract actual delta from
remaining_timeout) instead of subtracting a fixed 100, and after receiving
events send the appropriate acknowledgement back to the subscription (use the
subscription_pid passed into collect_with_logging to ack the batch or each event
so the subscriber buffer can free up) before recursing; keep the same function
names (collect_with_logging, subscription_pid, remaining_timeout) so the fix is
easy to locate.
| :expected -> | ||
| data = enqueue_events(data, events) | ||
| next_state(:max_capacity, data) | ||
| end |
There was a problem hiding this comment.
Queue grows unbounded beyond max_size in max_capacity
Medium Severity
The new notify_events handler in the max_capacity state calls enqueue_events(data, events) for :expected events. Previously, max_capacity had no explicit notify_events handler, so the catch-all at line 296 was used, which only called track_last_received without enqueuing. This means the internal queue can now grow well beyond the configured max_size limit if events keep arriving while the subscriber is slow to ack. With a fast producer and slow consumer, this could lead to unbounded memory growth — a regression from the old behavior where the queue was bounded by max_size and missed events were re-read from storage during catch-up.
| next_state(:max_capacity, data) | ||
|
|
||
| :future -> | ||
| next_state(:request_catch_up, data) |
There was a problem hiding this comment.
last_received not tracked for future events in max_capacity
Low Severity
The new max_capacity notify_events handler does not call track_last_received for the :future branch. The old catch-all handler (line 296–297) always called track_last_received(data, events) regardless of event continuity. Without this update, last_received may be stale when catch_up_from_stream checks last_sent == last_received to decide if catch-up is complete, potentially causing premature transition to :subscribed and delaying delivery of events that were notified via PubSub during max_capacity.
4 participants
No description provided.