Different behavior with `Buffer.take_count_or_until_permanent/2` when Buffer is size `:infinity`

[acco]

When a Buffer is any size < :infinity, Buffer.take_count_or_until_permanent/2 can return many permanents at a wheel position if they exist there:

alias GenStage.Buffer

# Create a buffer and store some temporary events
buffer = Buffer.new(10)
{buffer, _excess, _perms} = Buffer.store_temporary(buffer, [:temp1], :first)

# Store multiple permanent events at the same position
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm_first)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm_second)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm_third)

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

IO.inspect(temps)  # [:temp1] 
IO.inspect(perms)  # [:perm_first, :perm_second, :perm_third] ✓ Returns all three

However, when it is :infinity, it always stops at the first perm:

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

{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm_first)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm_second)
{:ok, buffer} = Buffer.store_permanent_unless_empty(buffer, :perm_third)

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

IO.inspect(perms) # [:perm_first] ❌  Only one perm returned

I assume we want these to have the same behavior?

[josevalim]

Yes, we can preserve the optimized behavior so we can send the several messages at once. :)