Skip to content

Fix permanent event batching for infinite buffers #325

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
josevalim merged 3 commits into from
Aug 5, 2026
+80 −1
Merged

Fix permanent event batching for infinite buffers #325

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7163e7c
Fix permanent event batching for infinite buffers
pckrishnadas88 Aug 4, 2026
a0f9028
Test infinite buffer count after taking permanents
pckrishnadas88 Aug 4, 2026
6ccf536
Handle permanents when infinite buffer counter reaches zero
pckrishnadas88 Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 23 additions & 1 deletion lib/gen_stage/buffer.ex
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,14 @@ defmodule GenStage.Buffer do
take_count_or_until_permanent(counter, [], queue, buffer, infos)
end

defp take_count_or_until_permanent(0, temps, queue, buffer, infos)
when is_reference(infos) do
{queue, buffer, perms} =
take_permanents(queue, buffer, infos, [])

{:ok, {queue, buffer, infos}, 0, :lists.reverse(temps), :lists.reverse(perms)}
end

defp take_count_or_until_permanent(0, temps, queue, buffer, infos) do
{:ok, {queue, buffer, infos}, 0, :lists.reverse(temps), []}
end
Expand All @@ -136,7 +144,10 @@ defmodule GenStage.Buffer do

case value do
{^infos, perm} ->
{:ok, {queue, buffer - 1, infos}, counter, :lists.reverse(temps), [perm]}
{queue, buffer, perms} =
take_permanents(queue, buffer - 1, infos, [perm])

{:ok, {queue, buffer, infos}, counter, :lists.reverse(temps), :lists.reverse(perms)}

@josevalim josevalim Aug 4, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What about the other clauses? Is this an issue:

When the last requested temporary event makes counter - 1 == 0, recursion enters the zero-counter clause before examining the following permanent entries and take_permanents/4 is never called:

buffer = Buffer.new(:infinity)
{buffer, _, _} = Buffer.store_temporary(buffer, [:temp], :first)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm1)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm2)

Buffer.take_count_or_until_permanent(buffer, 1)
#=> returns temps: [:temp], perms: [] and both permanents remain buffered

If so, we need to add a test for it.

@pckrishnadas88 pckrishnadas88 Aug 4, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, that was an issue. I added a regression test for this case and updated the infinite-buffer path to handle permanents when the counter reaches zero.


temp ->
take_count_or_until_permanent(counter - 1, [temp | temps], queue, buffer - 1, infos)
Expand All @@ -155,6 +166,17 @@ defmodule GenStage.Buffer do
end
end

defp take_permanents(queue, buffer, infos, perms) do
case :queue.peek(queue) do
{:value, {^infos, perm}} ->
{{:value, _}, queue} = :queue.out(queue)
take_permanents(queue, buffer - 1, infos, [perm | perms])

@josevalim josevalim Aug 4, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to worry about buffer going eventually negative? Can we add a test or is it bound to the queue size anyway?

@pckrishnadas88 pckrishnadas88 Aug 4, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe it is bounded by the queue size, since each permanent stored in the infinite buffer increments the count and the helper only decrements it when another permanent entry is actually present in the queue. I also added an assertion that the buffer count reaches 0 when all entries are consumed.


_ ->
{queue, buffer, perms}
end
end

## Wheel helpers

defp init_wheel(:infinity), do: make_ref()
Expand Down
57 changes: 57 additions & 0 deletions test/gen_stage/buffer_test.exs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,5 +132,62 @@ defmodule GenStage.BufferTest do
assert temps3 == [:temp10]
assert perms3 == [:perm11]
end

test "maintains FIFO order for permanent events with infinite buffer" do
buffer = Buffer.new(:infinity)

{buffer, _excess, _perms} =
Buffer.store_temporary(buffer, [:temp1, :temp2], :first)

{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm3)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm4)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm5)

{:ok, buffer, _remaining_count, temps, perms} =
Buffer.take_count_or_until_permanent(buffer, 5)

assert temps == [:temp1, :temp2]
assert perms == [:perm3, :perm4, :perm5]
assert Buffer.estimate_size(buffer) == 0
end

test "infinite buffer stops collecting permanents before next temporary event" do
buffer = Buffer.new(:infinity)

{buffer, _, _} = Buffer.store_temporary(buffer, [:temp1], :first)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm2)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm3)

{buffer, _, _} = Buffer.store_temporary(buffer, [:temp4], :first)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm5)

{:ok, buffer, remaining, temps, perms} =
Buffer.take_count_or_until_permanent(buffer, 5)

assert temps == [:temp1]
assert perms == [:perm2, :perm3]
assert remaining == 4

{:ok, _buffer, remaining, temps, perms} =
Buffer.take_count_or_until_permanent(buffer, remaining)

assert temps == [:temp4]
assert perms == [:perm5]
assert remaining == 3
end

test "infinite buffer returns permanents after the last requested temporary event" do
buffer = Buffer.new(:infinity)
{buffer, _, _} = Buffer.store_temporary(buffer, [:temp], :first)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm1)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm2)

{:ok, _buffer, remaining, temps, perms} =
Buffer.take_count_or_until_permanent(buffer, 1)

assert remaining == 0
assert temps == [:temp]
assert perms == [:perm1, :perm2]
end
end
end
Loading