Skip to content

ChannelSink.AsReader drops the final element when the consumer backpressures #8285

Description

@beminee

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.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

      Milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions