Akka.Streams: build island-boundary subscribers without runtime generics (Native AOT)

src/aot/Akka.Hosting.AOT.App/Program.cs

@@ -124,6 +124,9 @@
     await AssertDeathWatchAsync(system, host.Services);
     await AssertHealthAsync(host.Services);
 
+    // Akka.Streams graphs that cross island boundaries (#8731) - see StreamsScenarios.cs.
+    await StreamsScenarios.RunAsync(system);
+
     system.Log.Info(RoundTripMarker);
 
     // Proves fix for the race between Log.Info above and the watchdog's queue: the logger actor

src/aot/Akka.Hosting.AOT.App/README.md

@@ -5,8 +5,13 @@ through `Host.CreateApplicationBuilder` + `AddAkka(...)` and proves, against a r
 publish: a custom `WithExtension<T>` was registered by boot, not lazily created; a plain actor and a
 DI-constructed one both round-trip through `Ask`; `WithActorSystemLivenessCheck()` reports through
 `Microsoft.Extensions.Diagnostics.HealthChecks`; `ConfigureLoggers`'s `AddLoggerFactory` actually
-routes Akka's logs through `Microsoft.Extensions.Logging`; and `Watch`/`Terminated` (DeathWatch)
-works. A `WatchdogLoggerProvider` fails the run on any Warning/Error logged anywhere in the host.
+routes Akka's logs through `Microsoft.Extensions.Logging`; `Watch`/`Terminated` (DeathWatch)
+works; and Akka.Streams graphs that cross island boundaries run (`StreamsScenarios.cs`: `.Async()`,
+`Sink.AsPublisher` into `Source.FromPublisher`, a fan-out publisher, `Source.ActorPublisher`,
+`Source.Queue` and a graph that reads its own materialized value, each with a value-type and/or a
+reference-type element). The streams part needs no `TrimmerRootDescriptor`; if Akka.Streams ever
+builds a boundary type reflectively again, the run fails with `MissingMethodException` (#8731). A
+`WatchdogLoggerProvider` fails the run on any Warning/Error logged anywhere in the host.
 
 The app terminates itself once every assertion passes (`OK`, exit 0) instead of waiting on
 `host.WaitForShutdownAsync()`, so CI gets a deterministic exit.
@@ -27,6 +32,7 @@ with the AOT switch off.
 
 `aot-warnings.baseline.txt` in this directory tracks known `IL2xxx`/`IL3xxx` warnings under
 `src/contrib/hosting/`, `src/contrib/dependencyinjection/`, and `src/core/Akka.Streams/` (Hosting
-registers the stream-ref serializer at startup; see #8667). `src/core/Akka/` is out of scope - the
+registers the stream-ref serializer at startup; see #8667, and the streams scenarios reach the
+materializer's island-boundary code; see #8731). `src/core/Akka/` is out of scope - the
 plain-core canary's own baseline already covers it. Checked by the Hosting steps of the `AotCanary` job in
 `build-system/pr-validation.yaml`, which also documents how to reproduce the check locally.

src/aot/Akka.Hosting.AOT.App/StreamsScenarios.cs

@@ -0,0 +1,134 @@
+//-----------------------------------------------------------------------
+// <copyright file="StreamsScenarios.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2025 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2025 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System.Collections.Immutable;
+using Akka.Actor;
+using Akka.Streams;
+using Akka.Streams.Actors;
+using Akka.Streams.Dsl;
+
+namespace Akka.Hosting.AOT.App;
+
+/// <summary>
+/// Akka.Streams graphs that cross island boundaries (#8731). Each boundary used to be built with
+/// MakeGenericType + Activator, which Native AOT trims away. Every scenario runs once with a value
+/// type and once with a reference type, because the two get different compiled instantiations.
+/// No TrimmerRootDescriptor is involved: if a boundary type is still built reflectively, its
+/// constructor is gone and the run fails with MissingMethodException.
+/// </summary>
+internal static class StreamsScenarios
+{
+    private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10);
+
+    public static async Task RunAsync(ActorSystem system)
+    {
+        var mat = system.Materializer(namePrefix: "aot-canary");
+
+        // .Async() splits the graph into islands: BoundarySubscriber, BoundaryPublisher and
+        // ActorOutputBoundary on every edge between them.
+        await ExpectAsync(
+            Source.From(Enumerable.Range(1, 10)).Async().Select(x => x * 2).Async().RunWith(Sink.Seq<int>(), mat),
+            Enumerable.Range(1, 10).Select(x => x * 2), "async boundaries (int)");
+        await ExpectAsync(
+            Source.From(Enumerable.Range(1, 10)).Select(x => x.ToString()).Async().Select(s => s + "!").Async()
+                .RunWith(Sink.Seq<string>(), mat),
+            Enumerable.Range(1, 10).Select(x => x + "!"), "async boundaries (string)");
+
+        // Sink.AsPublisher(false) is a sink module that hands back a plain ISubscriber<T>, and
+        // Source.FromPublisher is a source module - both are wired up by MaterializerSession.
+        await ExpectAsync(
+            Source.FromPublisher(Source.From(Enumerable.Range(1, 5)).RunWith(Sink.AsPublisher<int>(false), mat))
+                .RunWith(Sink.Seq<int>(), mat),
+            Enumerable.Range(1, 5), "AsPublisher -> FromPublisher (int)");
+        await ExpectAsync(
+            Source.FromPublisher(Source.From(new[] { "a", "b", "c" }).RunWith(Sink.AsPublisher<string>(false), mat))
+                .RunWith(Sink.Seq<string>(), mat),
+            new[] { "a", "b", "c" }, "AsPublisher -> FromPublisher (string)");
+
+        // Sink.AsPublisher(true) runs a fan-out processor actor, which hands out ActorSubscriptions.
+        await ExpectAsync(
+            Source.FromPublisher(Source.From(Enumerable.Range(1, 5)).RunWith(Sink.AsPublisher<int>(true), mat))
+                .RunWith(Sink.Seq<int>(), mat),
+            Enumerable.Range(1, 5), "fan-out AsPublisher (int)");
+
+        // Source.ActorPublisher is what the persistence query read journals are built on.
+        await ExpectAsync(
+            Source.ActorPublisher<int>(Props.Create(() => new CountingPublisher<int>(5, i => i)))
+                .RunWith(Sink.Seq<int>(), mat),
+            Enumerable.Range(1, 5), "Source.ActorPublisher (int)");
+        await ExpectAsync(
+            Source.ActorPublisher<string>(Props.Create(() => new CountingPublisher<string>(3, i => "e" + i)))
+                .RunWith(Sink.Seq<string>(), mat),
+            new[] { "e1", "e2", "e3" }, "Source.ActorPublisher (string)");
+
+        // Source.Queue materializes a queue whose consumer sits behind an async boundary.
+        var (queue, done) = Source.Queue<int>(16, OverflowStrategy.Backpressure)
+            .Async()
+            .ToMaterialized(Sink.Seq<int>(), Keep.Both)
+            .Run(mat);
+        foreach (var i in Enumerable.Range(1, 3))
+            await queue.OfferAsync(i);
+        queue.Complete();
+        await ExpectAsync(done, new[] { 1, 2, 3 }, "Source.Queue (int)");
+
+        // A graph that reads its own materialized value (an int) back as a stream element:
+        // auto-fusing rebuilds the MaterializedValueSource it finds inside.
+        var seen = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var matValue = RunnableGraph.FromGraph(GraphDsl.Create(
+            Source.Single(1).MapMaterializedValue(_ => 42), (b, single) =>
+            {
+                b.From(single).To(Sink.Ignore<int>().MapMaterializedValue(_ => NotUsed.Instance));
+                b.From(b.MaterializedValue).To(Sink.ForEach<int>(x => seen.TrySetResult(x)).MapMaterializedValue(_ => NotUsed.Instance));
+                return ClosedShape.Instance;
+            })).Run(mat);
+        var echoed = await seen.Task.WaitAsync(Timeout);
+        if (matValue != 42 || echoed != 42)
+            throw new InvalidOperationException($"materialized value graph gave {matValue}/{echoed}, expected 42/42");
+    }
+
+    private static async Task ExpectAsync<T>(Task<IImmutableList<T>> run, IEnumerable<T> expected, string scenario)
+    {
+        var actual = await run.WaitAsync(Timeout);
+        if (!actual.SequenceEqual(expected))
+            throw new InvalidOperationException(
+                $"streams scenario '{scenario}' produced [{string.Join(", ", actual)}], expected [{string.Join(", ", expected)}]");
+    }
+
+    /// <summary>
+    /// Emits <c>1..count</c> (mapped to <typeparamref name="T"/>) as demand arrives, then completes.
+    /// </summary>
+    private sealed class CountingPublisher<T> : ActorPublisher<T>
+    {
+        private readonly int _count;
+        private readonly Func<int, T> _map;
+        private int _next = 1;
+
+        public CountingPublisher(int count, Func<int, T> map)
+        {
+            _count = count;
+            _map = map;
+        }
+
+        protected override bool Receive(object message)
+        {
+            switch (message)
+            {
+                case Request:
+                    while (TotalDemand > 0 && _next <= _count)
+                        OnNext(_map(_next++));
+                    if (_next > _count)
+                        OnCompleteThenStop();
+                    return true;
+                case Cancel:
+                    Context.Stop(Self);
+                    return true;
+                default:
+                    return false;
+            }
+        }
+    }
+}

src/aot/Akka.Hosting.AOT.App/aot-warnings.baseline.txt

@@ -7,11 +7,13 @@
 #   src/contrib/hosting/             - Akka.Hosting itself
 #   src/contrib/dependencyinjection/ - Akka.DependencyInjection
 #   src/core/Akka.Streams/           - not covered by the core canary, which never references
-#                                       Akka.Streams. Hosting does, and registers the stream-ref
-#                                       serializer unconditionally at startup.
+#                                       Akka.Streams. Hosting does, registers the stream-ref
+#                                       serializer unconditionally at startup, and the canary runs
+#                                       streams across island boundaries (StreamsScenarios.cs).
 #
 # src/core/Akka/ is deliberately OUT of scope - the plain-core canary's baseline already covers it.
 #
-# Empty since #8671 put the Akka.Streams stream-ref reflection behind Akka.DynamicTypeLoading. Any
+# Empty since #8671 put the Akka.Streams stream-ref reflection behind Akka.DynamicTypeLoading, and
+# #8731 stopped the materializer building island-boundary types with MakeGenericType. Any
 # new entry is a place where Hosting's dependency surface reaches for a type by name, or builds
 # one reflectively, in a way the trimmer cannot see.

src/core/Akka.Streams.Tests/Implementation/MaterializerSessionSpec.cs

@@ -0,0 +1,142 @@
+//-----------------------------------------------------------------------
+// <copyright file="MaterializerSessionSpec.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2025 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Collections.Immutable;
+using System.Linq;
+using System.Threading.Tasks;
+using Akka.Streams.Dsl;
+using Akka.Streams.Implementation;
+using Akka.Streams.TestKit;
+using Akka.TestKit;
+using FluentAssertions;
+using Xunit;
+
+namespace Akka.Streams.Tests.Implementation
+{
+    /// <summary>
+    /// Wiring and materialization-panic paths of <see cref="MaterializerSession"/>. These used to build
+    /// their wrappers with MakeGenericType from the runtime type of the subscriber or publisher; they
+    /// now use the element type the wrappers already carry (#8731).
+    /// </summary>
+    public class MaterializerSessionSpec : Akka.TestKit.Xunit.TestKit
+    {
+        public MaterializerSessionSpec(ITestOutputHelper output) : base(output: output)
+        {
+        }
+
+        [Fact(DisplayName = "Should_subscribe_a_plain_ISubscriber_through_the_publisher_When_a_sink_hands_one_over")]
+        public async Task Should_subscribe_a_plain_ISubscriber_through_the_publisher_When_a_sink_hands_one_over()
+        {
+            var publisher = this.CreateManualPublisherProbe<int>();
+            var subscriber = this.CreateManualSubscriberProbe<int>();
+            var session = new ProbeSession<int>(SourceToSink(), publisher, subscriber, plainSubscriber: true, throwAfter: int.MaxValue);
+
+            session.Materialize();
+
+            var subscription = await publisher.ExpectSubscriptionAsync();
+            subscription.Subscriber.Should().BeSameAs(subscriber);
+            await subscriber.ExpectSubscriptionAsync();
+        }
+
+        [Fact(DisplayName = "Should_fail_subscribers_and_cancel_publishers_When_materialization_panics_with_a_value_type")]
+        public Task Should_fail_subscribers_and_cancel_publishers_When_materialization_panics_with_a_value_type()
+            => PanicAsync<int>();
+
+        [Fact(DisplayName = "Should_fail_subscribers_and_cancel_publishers_When_materialization_panics_with_a_reference_type")]
+        public Task Should_fail_subscribers_and_cancel_publishers_When_materialization_panics_with_a_reference_type()
+            => PanicAsync<string>();
+
+        private async Task PanicAsync<T>()
+        {
+            var publisher = this.CreateManualPublisherProbe<T>();
+            var subscriber = this.CreateManualSubscriberProbe<T>();
+            var session = new ProbeSession<T>(SourceToSink(), publisher, subscriber, plainSubscriber: false, throwAfter: 2);
+
+            session.Invoking(s => s.Materialize()).Should().Throw<TestException>().WithMessage("boom");
+
+            // the regular wiring happened before the panic
+            (await publisher.ExpectSubscriptionAsync()).Subscriber.Should().BeSameAs(subscriber);
+            await subscriber.ExpectSubscriptionAsync();
+
+            // the panic subscribes a CancellingSubscriber<T> to every publisher...
+            var cancelling = await publisher.ExpectSubscriptionAsync();
+            cancelling.Subscriber.Should().BeOfType<CancellingSubscriber<T>>();
+            await cancelling.ExpectCancellationAsync();
+
+            // ...and an ErrorPublisher<T> to every subscriber
+            await subscriber.ExpectSubscriptionAsync();
+            var error = await subscriber.ExpectErrorAsync();
+            error.Should().BeOfType<MaterializerSession.MaterializationPanicException>()
+                .Which.InnerException.Should().BeOfType<TestException>();
+        }
+
+        private static IModule SourceToSink()
+        {
+            var source = new ProbeModule(inlets: 0, outlets: 1);
+            var sink = new ProbeModule(inlets: 1, outlets: 0);
+            return source
+                .Compose<object, object, object>(sink, Keep.None)
+                .Wire(source.OutPorts.First(), sink.InPorts.First());
+        }
+
+        private sealed class ProbeModule : AtomicModule
+        {
+            public ProbeModule(int inlets, int outlets)
+            {
+                Shape = new AmorphousShape(
+                    Enumerable.Range(0, inlets).Select(i => new Inlet<object>("in" + i)).ToImmutableArray<Inlet>(),
+                    Enumerable.Range(0, outlets).Select(i => new Outlet<object>("out" + i)).ToImmutableArray<Outlet>());
+            }
+
+            public override Shape Shape { get; }
+            public override IModule ReplaceShape(Shape shape) => throw new NotSupportedException();
+            public override IModule CarbonCopy() => throw new NotSupportedException();
+            public override Attributes Attributes => Attributes.None;
+            public override IModule WithAttributes(Attributes attributes) => this;
+        }
+
+        /// <summary>
+        /// Assigns the probes to the ports of each atomic module, the way a source or sink module would,
+        /// and throws once <c>throwAfter</c> atomic modules have been materialized.
+        /// </summary>
+        private sealed class ProbeSession<T> : MaterializerSession
+        {
+            private readonly TestPublisher.ManualProbe<T> _publisher;
+            private readonly TestSubscriber.ManualProbe<T> _subscriber;
+            private readonly bool _plainSubscriber;
+            private readonly int _throwAfter;
+            private int _materialized;
+
+            public ProbeSession(IModule module, TestPublisher.ManualProbe<T> publisher, TestSubscriber.ManualProbe<T> subscriber,
+                bool plainSubscriber, int throwAfter) : base(module, Attributes.None)
+            {
+                _publisher = publisher;
+                _subscriber = subscriber;
+                _plainSubscriber = plainSubscriber;
+                _throwAfter = throwAfter;
+            }
+
+            protected override object MaterializeAtomic(AtomicModule atomic, Attributes effectiveAttributes,
+                IDictionary<IModule, object> materializedValues)
+            {
+                foreach (var inPort in atomic.InPorts)
+                    AssignPort(inPort, _plainSubscriber ? _subscriber : UntypedSubscriber.FromTyped(_subscriber));
+                foreach (var outPort in atomic.OutPorts)
+                    AssignPort(outPort, UntypedPublisher.FromTyped(_publisher));
+
+                if (++_materialized == _throwAfter)
+                    throw new TestException("boom");
+
+                return NotUsed.Instance;
+            }
+        }
+    }
+}