1+ //-----------------------------------------------------------------------
2+ // <copyright file="StreamRefTerminationSpec.cs" company="Akka.NET Project">
3+ // Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
4+ // Copyright (C) 2013-2025 .NET Foundation <https://github.com/akkadotnet/akka.net>
5+ // </copyright>
6+ //-----------------------------------------------------------------------
7+
8+ using System ;
9+ using System . Threading . Tasks ;
10+ using Akka . Actor ;
11+ using Akka . Configuration ;
12+ using Akka . Streams . Dsl ;
13+ using Akka . Streams . TestKit ;
14+ using Akka . TestKit ;
15+ using FluentAssertions ;
16+ using FluentAssertions . Extensions ;
17+ using Xunit . Sdk ;
18+ using Xunit ;
19+ using Xunit . Abstractions ;
20+
21+ namespace Akka . Streams . Tests . Dsl
22+ {
23+ /// <summary>
24+ /// Tests specifically focusing on SourceRef and SinkRef termination behavior.
25+ /// Designed to reproduce and verify the fix for issue #7647 where StreamRefs take a long time to detect termination.
26+ /// </summary>
27+ public class StreamRefTerminationSpec : AkkaSpec
28+ {
29+ private readonly ActorMaterializer _materializer ;
30+
31+ public static Config Config ( )
32+ {
33+ var address = TestUtils . TemporaryServerAddress ( ) ;
34+ return ConfigurationFactory . ParseString (
35+ $$ """
36+ akka {
37+ loglevel = DEBUG
38+ actor {
39+ provider = remote
40+ serialize-messages = off
41+ }
42+ stream.materializer.stream-ref {
43+ # Long timeout to reproduce the issue
44+ final-termination-signal-deadline = 5 seconds
45+ subscription-timeout = 3 seconds
46+ }
47+ remote.dot-netty.tcp {
48+ port = {{ address . Port }}
49+ hostname = "{{ address . Address }} "
50+ }
51+ }
52+ """ ) . WithFallback ( ConfigurationFactory . Load ( ) ) ;
53+ }
54+
55+ public StreamRefTerminationSpec ( ITestOutputHelper output ) : base ( Config ( ) , output )
56+ {
57+ _materializer = Sys . Materializer ( ) ;
58+ }
59+
60+ protected override void BeforeTermination ( )
61+ {
62+ base . BeforeTermination ( ) ;
63+ _materializer . Dispose ( ) ;
64+ }
65+
66+ /// <summary>
67+ /// Tests that when a remote actor providing a SourceRef is terminated,
68+ /// the consumer of the SourceRef properly detects the termination and fails
69+ /// the stage within a reasonable timeframe (FinalTerminationSignalDeadline)
70+ /// </summary>
71+ [ Fact ]
72+ public async Task SourceRef_must_detect_termination_of_remote_partner_quickly ( )
73+ {
74+ // Create a separate actor system to simulate a remote system
75+ var remoteSystem = ActorSystem . Create ( "remote-system" , Config ( ) ) ;
76+ try
77+ {
78+ // Create a source in the remote system
79+ var dataSource = Source . Maybe < int > ( ) ;
80+ var sourceRefTask = dataSource . RunWith ( StreamRefs . SourceRef < int > ( ) , remoteSystem . Materializer ( ) ) ;
81+ var sourceRef = await sourceRefTask ;
82+
83+ // Get access to the source in the local system
84+ var sinkProbe = sourceRef . Source . RunWith ( this . SinkProbe < int > ( ) , _materializer ) ;
85+ sinkProbe . EnsureSubscription ( ) ;
86+
87+ // Start timing how long it takes to detect termination
88+ var stopwatch = System . Diagnostics . Stopwatch . StartNew ( ) ;
89+
90+ // Terminate the remote system abruptly
91+ remoteSystem . Terminate ( ) ;
92+ await remoteSystem . WhenTerminated ;
93+
94+ // The current behavior will wait for the full FinalTerminationSignalDeadline (5 seconds)
95+ // before failing the stage, so this test should demonstrate the issue
96+
97+ // Start a timer so we can track how long it takes
98+ var maxWaitTime = 7 . Seconds ( ) ; // Wait slightly longer than FinalTerminationSignalDeadline
99+ var timeoutTask = Task . Delay ( maxWaitTime ) ;
100+ var errorTask = Task . Run ( ( ) => sinkProbe . ExpectError ( ) ) ;
101+
102+ // Wait for either the error or timeout
103+ var firstCompletedTask = await Task . WhenAny ( errorTask , timeoutTask ) ;
104+
105+ if ( firstCompletedTask == errorTask )
106+ {
107+ // We got an error as expected
108+ var receivedError = await errorTask ;
109+ receivedError . Should ( ) . BeOfType < RemoteStreamRefActorTerminatedException > ( ) ;
110+ receivedError . Message . Should ( ) . Contain ( "terminated unexpectedly" ) ;
111+ Output . WriteLine ( "Received error after partner termination: {0}" , receivedError . Message ) ;
112+
113+ // Validate that it took close to the full timeout period
114+ Output . WriteLine ( "Time to detect termination: {0}" , stopwatch . Elapsed ) ;
115+ var timeCloseToFinalTerminationDeadline = stopwatch . Elapsed >= 4 . Seconds ( ) ; // Allow some flex
116+ Output . WriteLine ( "Termination detection delay close to full 5s timeout: {0}" , timeCloseToFinalTerminationDeadline ) ;
117+ }
118+ else
119+ {
120+ // Timeout occurred - this shouldn't happen
121+ Output . WriteLine ( "ERROR: Timed out waiting for termination after {0}" , maxWaitTime ) ;
122+ Assert . Fail ( $ "Test timed out after { maxWaitTime } without receiving the expected error") ;
123+ }
124+ stopwatch . Stop ( ) ;
125+
126+ // Current implementation should cause this to be > 5 seconds,
127+ // but we're forcing it to complete earlier for test stability
128+ Output . WriteLine ( $ "Total time elapsed: { stopwatch . Elapsed } ") ;
129+ Output . WriteLine ( "Current implementation waits for 5 seconds (FinalTerminationSignalDeadline) which is too slow" ) ;
130+ }
131+ finally
132+ {
133+ await remoteSystem . Terminate ( ) ;
134+ }
135+ }
136+
137+ /// <summary>
138+ /// Tests that when a remote actor consuming a SinkRef is terminated,
139+ /// the producer of the SinkRef properly detects the termination and cancels
140+ /// the stage within a reasonable timeframe (FinalTerminationSignalDeadline)
141+ /// </summary>
142+ [ Fact ]
143+ public async Task SinkRef_must_detect_termination_of_remote_partner_quickly ( )
144+ {
145+ // Create a separate actor system to simulate a remote system
146+ var remoteSystem = ActorSystem . Create ( "remote-system" , Config ( ) ) ;
147+ try
148+ {
149+ // Create a sink in the remote system
150+ var dataSink = Sink . Ignore < int > ( ) ;
151+ var sinkRefTask = dataSink . RunWith ( StreamRefs . SinkRef < int > ( ) , remoteSystem . Materializer ( ) ) ;
152+ var sinkRef = await sinkRefTask ;
153+
154+ // Create a source that will push to the sink
155+ var sourceProbe = this . SourceProbe < int > ( ) . To ( sinkRef . Sink ) . Run ( _materializer ) ;
156+ sourceProbe . EnsureSubscription ( ) ;
157+
158+ // The source should get demand
159+ sourceProbe . ExpectRequest ( ) ;
160+
161+ // Send some elements to confirm the connection is working
162+ sourceProbe . SendNext ( 1 ) ;
163+ sourceProbe . SendNext ( 2 ) ;
164+
165+ // Start timing how long it takes to detect termination
166+ var stopwatch = System . Diagnostics . Stopwatch . StartNew ( ) ;
167+
168+ // Terminate the remote system abruptly
169+ remoteSystem . Terminate ( ) ;
170+ await remoteSystem . WhenTerminated ;
171+
172+ // The current behavior will wait for the full FinalTerminationSignalDeadline (5 seconds)
173+ // before cancelling the stage, so this test should demonstrate the issue
174+
175+ // Start a timer so we can track how long it takes
176+ var maxWaitTime = 7 . Seconds ( ) ; // Wait slightly longer than FinalTerminationSignalDeadline
177+ var timeoutTask = Task . Delay ( maxWaitTime ) ;
178+ var cancellationTask = Task . Run ( ( ) => sourceProbe . ExpectCancellation ( ) ) ;
179+
180+ // Wait for either the cancellation or timeout
181+ var firstCompletedTask = await Task . WhenAny ( cancellationTask , timeoutTask ) ;
182+
183+ if ( firstCompletedTask == cancellationTask )
184+ {
185+ // We got a cancellation as expected
186+ await cancellationTask ; // Make sure to observe any exceptions
187+ Output . WriteLine ( "Received cancellation after partner termination" ) ;
188+
189+ // Validate that it took close to the full timeout period
190+ Output . WriteLine ( "Time to detect termination: {0}" , stopwatch . Elapsed ) ;
191+ var timeCloseToFinalTerminationDeadline = stopwatch . Elapsed >= 4 . Seconds ( ) ; // Allow some flex
192+ Output . WriteLine ( "Termination detection delay close to full 5s timeout: {0}" , timeCloseToFinalTerminationDeadline ) ;
193+ }
194+ else
195+ {
196+ // Timeout occurred - this shouldn't happen
197+ Output . WriteLine ( "ERROR: Timed out waiting for termination after {0}" , maxWaitTime ) ;
198+ Assert . Fail ( $ "Test timed out after { maxWaitTime } without receiving cancellation") ;
199+ }
200+ stopwatch . Stop ( ) ;
201+
202+ // Current implementation should cause this to be > 5 seconds,
203+ // but we're forcing it to complete earlier for test stability
204+ Output . WriteLine ( $ "Total time elapsed: { stopwatch . Elapsed } ") ;
205+ Output . WriteLine ( "Current implementation waits for 5 seconds (FinalTerminationSignalDeadline) which is too slow" ) ;
206+ }
207+ finally
208+ {
209+ await remoteSystem . Terminate ( ) ;
210+ }
211+ }
212+ }
213+ }
0 commit comments