Version Information
Akka.Streams 1.5.60 (reproduced; likely affects other 1.5.x).
Describe the bug
When an Akka.Streams source is run into ChannelSink.AsReader<T>(bufferSize, …, BoundedChannelFullMode.Wait) and the resulting ChannelReader<T> is consumed more slowly than the source produces (so the bounded channel fills and the stage backpressures), the last element of the stream is silently lost. The reader completes one element short. With a consumer that keeps up (no backpressure), all elements are delivered correctly, so the bug only manifests under backpressure, which is exactly the scenario the operator exists for.
To Reproduce
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Channels;
using System.Threading.Tasks;
using Akka.Actor;
using Akka.Streams;
using Akka.Streams.Dsl;
using var system = ActorSystem.Create("repro");
var materializer = system.Materializer();
const int elementCount = 30;
const int bufferSize = 4;
// Run a source of 0..29 into ChannelSink.AsReader, then read the ChannelReader slowly.
ChannelReader<int> reader = Source
.From(Enumerable.Range(0, elementCount))
.RunWith(
ChannelSink.AsReader<int>(bufferSize, singleReader: true, BoundedChannelFullMode.Wait),
materializer);
var received = new List<int>();
await foreach (var item in reader.ReadAllAsync())
{
received.Add(item);
await Task.Delay(1); // slow consumer => bounded channel fills => stage backpressures
}
Console.WriteLine($"expected: {elementCount} elements, last = {elementCount - 1}");
Console.WriteLine($"received: {received.Count} elements, last = {received.LastOrDefault()}");
// Expected: received 30, last 29
// Actual: received 29, last 28 (final element 29 is missing)
await system.Terminate();
Expected behavior
received contains all 30 elements (0..29); the reader yields every element the source produced before completing.
Actual behavior
received contains 29 elements (0..28); element 29 (the last one the source emitted) is dropped. The ChannelReader then reports completion as if the stream ended normally. No error is surfaced.
Environment
Windows 11, .NET 10. Should be reproducible in every version.
Additional context
Deterministic under backpressure. In a 40-iteration loop of the snippet above it dropped the last element 40/40 times (always exactly the final element). Removing the await Task.Delay(1) makes the problem disappear.
Version Information
Akka.Streams 1.5.60 (reproduced; likely affects other 1.5.x).
Describe the bug
When an Akka.Streams source is run into
ChannelSink.AsReader<T>(bufferSize, …, BoundedChannelFullMode.Wait)and the resultingChannelReader<T>is consumed more slowly than the source produces (so the bounded channel fills and the stage backpressures), the last element of the stream is silently lost. The reader completes one element short. With a consumer that keeps up (no backpressure), all elements are delivered correctly, so the bug only manifests under backpressure, which is exactly the scenario the operator exists for.To Reproduce
Expected behavior
receivedcontains all 30 elements (0..29); the reader yields every element the source produced before completing.Actual behavior
receivedcontains 29 elements (0..28); element29(the last one the source emitted) is dropped. The ChannelReader then reports completion as if the stream ended normally. No error is surfaced.Environment
Windows 11, .NET 10. Should be reproducible in every version.
Additional context
Deterministic under backpressure. In a 40-iteration loop of the snippet above it dropped the last element 40/40 times (always exactly the final element). Removing the
await Task.Delay(1)makes the problem disappear.