From 85e50b9fb69fdfcd232d49e8c6c27876ddcf87d0 Mon Sep 17 00:00:00 2001 From: Javier Calvarro Nelson <6995051+javiercn@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:29:14 +0200 Subject: [PATCH 1/7] Reject duplicate SignalR upload stream IDs Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../Core/src/Internal/DefaultHubDispatcher.cs | 33 ++++---- src/SignalR/server/Core/src/StreamTracker.cs | 43 ++++++++-- .../HubConnectionHandlerTestUtils/Hubs.cs | 19 +++++ .../HubConnectionHandlerTests.cs | 82 +++++++++++++++++++ 4 files changed, 157 insertions(+), 20 deletions(-) diff --git a/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs b/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs index f049147911f7..4a2addcdd622 100644 --- a/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs +++ b/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs @@ -370,6 +370,7 @@ private async Task Invoke(HubMethodDescriptor descriptor, HubConnectionCon var scope = _serviceScopeFactory.CreateAsyncScope(); IHubActivator? hubActivator = null; THub? hub = null; + List? streamRegistrations = null; try { hubActivator = scope.ServiceProvider.GetRequiredService>(); @@ -409,7 +410,7 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, CancellationTokenSource? cts = null; if (descriptor.HasSyntheticArguments) { - ReplaceArguments(descriptor, hubMethodInvocationMessage, isStreamCall, connection, scope, ref arguments, out cts); + ReplaceArguments(descriptor, hubMethodInvocationMessage, isStreamCall, connection, scope, ref arguments, ref streamRegistrations, out cts); } if (isStreamCall || isStreamResponse) @@ -424,7 +425,7 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, if (isStreamResponse) { _ = StreamAsync(hubMethodInvocationMessage.InvocationId!, connection, hubCallerContext, - arguments, scope, hubActivator, hub, cts, hubMethodInvocationMessage, descriptor); + arguments, scope, hubActivator, hub, cts, hubMethodInvocationMessage, descriptor, streamRegistrations); } else { @@ -439,7 +440,8 @@ static async Task ExecuteInvocation(DefaultHubDispatcher dispatcher, HubCallerContext hubCallerContext, HubMethodInvocationMessage hubMethodInvocationMessage, bool isStreamCall, - CancellationTokenSource? cts) + CancellationTokenSource? cts, + List? streamRegistrations) { var logger = dispatcher._logger; var enableDetailedErrors = dispatcher._enableDetailedErrors; @@ -509,7 +511,7 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, // And normal invocations handle cleanup below in the finally if (isStreamCall) { - await CleanupInvocation(connection, hubMethodInvocationMessage, hubActivator, hub, scope); + await CleanupInvocation(connection, streamRegistrations, hubActivator, hub, scope); } } @@ -521,7 +523,7 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, } } - invocation = ExecuteInvocation(this, methodExecutor, hub, arguments, scope, hubActivator, connection, hubCallerContext, hubMethodInvocationMessage, isStreamCall, cts); + invocation = ExecuteInvocation(this, methodExecutor, hub, arguments, scope, hubActivator, connection, hubCallerContext, hubMethodInvocationMessage, isStreamCall, cts, streamRegistrations); } if (isStreamCall || isStreamResponse) @@ -557,21 +559,21 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, { wasSemaphoreReleased = !hubCallerClients.TrySetSemaphoreReleased(); } - await CleanupInvocation(connection, hubMethodInvocationMessage, hubActivator, hub, scope); + await CleanupInvocation(connection, streamRegistrations, hubActivator, hub, scope); } } return !wasSemaphoreReleased; } - private static ValueTask CleanupInvocation(HubConnectionContext connection, HubMethodInvocationMessage hubMessage, IHubActivator? hubActivator, + private static ValueTask CleanupInvocation(HubConnectionContext connection, List? streamRegistrations, IHubActivator? hubActivator, THub? hub, AsyncServiceScope scope) { - if (hubMessage.StreamIds != null) + if (streamRegistrations is not null) { - foreach (var stream in hubMessage.StreamIds) + foreach (var streamRegistration in streamRegistrations) { - connection.StreamTracker.TryComplete(CompletionMessage.Empty(stream)); + connection.StreamTracker.TryComplete(streamRegistration); } } @@ -584,7 +586,8 @@ private static ValueTask CleanupInvocation(HubConnectionContext connection, HubM } private async Task StreamAsync(string invocationId, HubConnectionContext connection, HubCallerContext hubCallerContext, object?[] arguments, AsyncServiceScope scope, - IHubActivator hubActivator, THub hub, CancellationTokenSource? streamCts, HubMethodInvocationMessage hubMethodInvocationMessage, HubMethodDescriptor descriptor) + IHubActivator hubActivator, THub hub, CancellationTokenSource? streamCts, HubMethodInvocationMessage hubMethodInvocationMessage, + HubMethodDescriptor descriptor, List? streamRegistrations) { string? error = null; @@ -671,7 +674,7 @@ private async Task StreamAsync(string invocationId, HubConnectionContext connect Activity.Current = previousActivity; } - await CleanupInvocation(connection, hubMethodInvocationMessage, hubActivator, hub, scope); + await CleanupInvocation(connection, streamRegistrations, hubActivator, hub, scope); // Only remove/dispose the CTS if we successfully registered it, otherwise we'd evict // another invocation's CTS on ID collision. @@ -817,7 +820,8 @@ await connection.WriteAsync(CompletionMessage.WithError(hubMethodInvocationMessa } private void ReplaceArguments(HubMethodDescriptor descriptor, HubMethodInvocationMessage hubMethodInvocationMessage, bool isStreamCall, - HubConnectionContext connection, AsyncServiceScope scope, ref object?[] arguments, out CancellationTokenSource? cts) + HubConnectionContext connection, AsyncServiceScope scope, ref object?[] arguments, + ref List? streamRegistrations, out CancellationTokenSource? cts) { cts = null; // In order to add the synthetic arguments we need a new array because the invocation array is too small (it doesn't know about synthetic arguments) @@ -842,7 +846,8 @@ private void ReplaceArguments(HubMethodDescriptor descriptor, HubMethodInvocatio Log.StartingParameterStream(_logger, hubMethodInvocationMessage.StreamIds![streamPointer]); var itemType = descriptor.StreamingParameters![streamPointer]; arguments[parameterPointer] = connection.StreamTracker.AddStream(hubMethodInvocationMessage.StreamIds[streamPointer], - itemType, descriptor.OriginalParameterTypes[parameterPointer]); + itemType, descriptor.OriginalParameterTypes[parameterPointer], out var streamRegistration); + (streamRegistrations ??= []).Add(streamRegistration); streamPointer++; } diff --git a/src/SignalR/server/Core/src/StreamTracker.cs b/src/SignalR/server/Core/src/StreamTracker.cs index 81fea34566de..156628a6778c 100644 --- a/src/SignalR/server/Core/src/StreamTracker.cs +++ b/src/SignalR/server/Core/src/StreamTracker.cs @@ -24,19 +24,25 @@ public StreamTracker(int streamBufferCapacity) } /// - /// Creates a new stream and returns the ChannelReader for it as an object. + /// Creates a new stream and returns the ChannelReader for it as an object and a registration used for cleanup. /// [UnconditionalSuppressMessage("Trimming", "IL2060:MakeGenericMethod", Justification = "BuildStream doesn't have trimming annotations.")] [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "HubMethodDescriptor checks for ValueType streaming item types when PublishAot=true. Developers will get an exception in this situation before publishing.")] - public object AddStream(string streamId, Type itemType, Type targetType) + public object AddStream(string streamId, Type itemType, Type targetType, out StreamRegistration streamRegistration) { Debug.Assert(RuntimeFeature.IsDynamicCodeSupported || !itemType.IsValueType, "HubMethodDescriptor ensures itemType is not a ValueType when PublishAot=true."); var newConverter = (IStreamConverter)_buildConverterMethod.MakeGenericMethod(itemType).Invoke(null, _streamConverterArgs)!; - _lookup[streamId] = newConverter; - return newConverter.GetReaderAsObject(targetType); + var reader = newConverter.GetReaderAsObject(targetType); + if (!_lookup.TryAdd(streamId, newConverter)) + { + throw new HubException($"Stream ID '{streamId}' is already in use."); + } + + streamRegistration = new StreamRegistration(streamId, newConverter); + return reader; } private bool TryGetConverter(string streamId, [NotNullWhen(true)] out IStreamConverter? converter) @@ -74,7 +80,7 @@ public Type GetStreamItemType(string streamId) public bool TryComplete(CompletionMessage message) { _lookup.TryRemove(message.InvocationId!, out var converter); - if (converter == null) + if (converter is null) { return false; } @@ -82,6 +88,18 @@ public bool TryComplete(CompletionMessage message) return true; } + public bool TryComplete(StreamRegistration streamRegistration) + { + var stream = new KeyValuePair(streamRegistration.StreamId, streamRegistration.Converter); + if (((ICollection>)_lookup).Remove(stream)) + { + streamRegistration.Converter.TryComplete(null); + return true; + } + + return false; + } + public void CompleteAll(Exception ex) { foreach (var converter in _lookup) @@ -95,7 +113,20 @@ private static IStreamConverter BuildStream(int streamBufferCapacity) return new ChannelConverter(streamBufferCapacity); } - private interface IStreamConverter + public readonly struct StreamRegistration + { + internal StreamRegistration(string streamId, IStreamConverter converter) + { + StreamId = streamId; + Converter = converter; + } + + internal string StreamId { get; } + + internal IStreamConverter Converter { get; } + } + + internal interface IStreamConverter { Type GetItemType(); object GetReaderAsObject(Type type); diff --git a/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTestUtils/Hubs.cs b/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTestUtils/Hubs.cs index afb56aa755a0..2e8fdbde2371 100644 --- a/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTestUtils/Hubs.cs +++ b/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTestUtils/Hubs.cs @@ -223,6 +223,25 @@ public async Task StreamingConcat(ChannelReader source) return sb.ToString(); } + public async Task StreamingConcatTwoStreams(ChannelReader first, ChannelReader second) + { + return await ReadStream(first) + await ReadStream(second); + + static async Task ReadStream(ChannelReader source) + { + var result = new StringBuilder(); + while (await source.WaitToReadAsync()) + { + while (source.TryRead(out var item)) + { + result.Append(item); + } + } + + return result.ToString(); + } + } + public async Task StreamDontRead(ChannelReader source) { while (await source.WaitToReadAsync()) diff --git a/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTests.cs b/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTests.cs index 051ae6409ac0..6fe265df7af5 100644 --- a/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTests.cs +++ b/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTests.cs @@ -3696,6 +3696,88 @@ public async Task UploadStringsToConcat() } } + [Fact] + public async Task UploadStreamWithDuplicateIdsFailsAndConnectionContinues() + { + var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services => + { + services.AddSignalR(options => + { + options.EnableDetailedErrors = true; + options.StreamBufferCapacity = 1; + }); + }); + var connectionHandler = serviceProvider.GetService>(); + + using (var client = new TestClient()) + { + var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout(); + await client.BeginUploadStreamAsync("duplicate", nameof(MethodHub.StreamingConcatTwoStreams), new[] { "id", "id" }, Array.Empty()).DefaultTimeout(); + await client.SendHubMessageAsync(new StreamItemMessage("id", "first")).DefaultTimeout(); + await client.SendHubMessageAsync(new StreamItemMessage("id", "second")).DefaultTimeout(); + await client.SendInvocationAsync(nameof(MethodHub.Echo), "test").DefaultTimeout(); + + var duplicateCompletion = Assert.IsType(await client.ReadAsync().DefaultTimeout()); + Assert.Equal("An unexpected error occurred invoking 'StreamingConcatTwoStreams' on the server. HubException: Stream ID 'id' is already in use.", duplicateCompletion.Error); + + var echoCompletion = Assert.IsType(await client.ReadAsync().DefaultTimeout()); + Assert.Equal("test", echoCompletion.Result); + } + } + + [Fact] + public async Task ActiveUploadStreamCannotBeReplacedAndIdCanBeReusedAfterCompletion() + { + var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services => + { + services.AddSignalR(options => options.EnableDetailedErrors = true); + }); + var connectionHandler = serviceProvider.GetService>(); + + using (var client = new TestClient()) + { + var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout(); + await client.BeginUploadStreamAsync("original", nameof(MethodHub.StreamingConcat), new[] { "id" }, Array.Empty()).DefaultTimeout(); + await client.BeginUploadStreamAsync("duplicate", nameof(MethodHub.StreamingConcat), new[] { "id" }, Array.Empty()).DefaultTimeout(); + + var duplicateCompletion = Assert.IsType(await client.ReadAsync().DefaultTimeout()); + Assert.Equal("An unexpected error occurred invoking 'StreamingConcat' on the server. HubException: Stream ID 'id' is already in use.", duplicateCompletion.Error); + + await client.SendHubMessageAsync(new StreamItemMessage("id", "original")).DefaultTimeout(); + await client.SendHubMessageAsync(CompletionMessage.Empty("id")).DefaultTimeout(); + + var originalCompletion = Assert.IsType(await client.ReadAsync().DefaultTimeout()); + Assert.Equal("original", originalCompletion.Result); + + await client.BeginUploadStreamAsync("reused", nameof(MethodHub.StreamingConcat), new[] { "id" }, Array.Empty()).DefaultTimeout(); + await client.SendHubMessageAsync(new StreamItemMessage("id", "reused")).DefaultTimeout(); + await client.SendHubMessageAsync(CompletionMessage.Empty("id")).DefaultTimeout(); + + var reusedCompletion = Assert.IsType(await client.ReadAsync().DefaultTimeout()); + Assert.Equal("reused", reusedCompletion.Result); + } + } + + [Fact] + public async Task UploadMultipleStreamsWithUniqueIds() + { + var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(); + var connectionHandler = serviceProvider.GetService>(); + + using (var client = new TestClient()) + { + var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout(); + await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.StreamingConcatTwoStreams), new[] { "first", "second" }, Array.Empty()).DefaultTimeout(); + await client.SendHubMessageAsync(new StreamItemMessage("first", "hello ")).DefaultTimeout(); + await client.SendHubMessageAsync(CompletionMessage.Empty("first")).DefaultTimeout(); + await client.SendHubMessageAsync(new StreamItemMessage("second", "world")).DefaultTimeout(); + await client.SendHubMessageAsync(CompletionMessage.Empty("second")).DefaultTimeout(); + + var completion = Assert.IsType(await client.ReadAsync().DefaultTimeout()); + Assert.Equal("hello world", completion.Result); + } + } + [Fact] public async Task UploadStreamedObjects() { From 28ff700c1ad6185b4d85e4ca25f4d796ba7f3fd2 Mon Sep 17 00:00:00 2001 From: Javier Calvarro Nelson <6995051+javiercn@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:29:14 +0200 Subject: [PATCH 2/7] Simplify upload stream ownership cleanup Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../Core/src/Internal/DefaultHubDispatcher.cs | 31 ++++++------ src/SignalR/server/Core/src/StreamTracker.cs | 50 +++++++++---------- 2 files changed, 40 insertions(+), 41 deletions(-) diff --git a/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs b/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs index 4a2addcdd622..3224ab4abf0f 100644 --- a/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs +++ b/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs @@ -370,7 +370,7 @@ private async Task Invoke(HubMethodDescriptor descriptor, HubConnectionCon var scope = _serviceScopeFactory.CreateAsyncScope(); IHubActivator? hubActivator = null; THub? hub = null; - List? streamRegistrations = null; + object? streamOwner = null; try { hubActivator = scope.ServiceProvider.GetRequiredService>(); @@ -410,7 +410,7 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, CancellationTokenSource? cts = null; if (descriptor.HasSyntheticArguments) { - ReplaceArguments(descriptor, hubMethodInvocationMessage, isStreamCall, connection, scope, ref arguments, ref streamRegistrations, out cts); + ReplaceArguments(descriptor, hubMethodInvocationMessage, isStreamCall, connection, scope, ref arguments, ref streamOwner, out cts); } if (isStreamCall || isStreamResponse) @@ -425,7 +425,7 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, if (isStreamResponse) { _ = StreamAsync(hubMethodInvocationMessage.InvocationId!, connection, hubCallerContext, - arguments, scope, hubActivator, hub, cts, hubMethodInvocationMessage, descriptor, streamRegistrations); + arguments, scope, hubActivator, hub, cts, hubMethodInvocationMessage, descriptor, streamOwner); } else { @@ -441,7 +441,7 @@ static async Task ExecuteInvocation(DefaultHubDispatcher dispatcher, HubMethodInvocationMessage hubMethodInvocationMessage, bool isStreamCall, CancellationTokenSource? cts, - List? streamRegistrations) + object? streamOwner) { var logger = dispatcher._logger; var enableDetailedErrors = dispatcher._enableDetailedErrors; @@ -511,7 +511,7 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, // And normal invocations handle cleanup below in the finally if (isStreamCall) { - await CleanupInvocation(connection, streamRegistrations, hubActivator, hub, scope); + await CleanupInvocation(connection, hubMethodInvocationMessage, streamOwner, hubActivator, hub, scope); } } @@ -523,7 +523,7 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, } } - invocation = ExecuteInvocation(this, methodExecutor, hub, arguments, scope, hubActivator, connection, hubCallerContext, hubMethodInvocationMessage, isStreamCall, cts, streamRegistrations); + invocation = ExecuteInvocation(this, methodExecutor, hub, arguments, scope, hubActivator, connection, hubCallerContext, hubMethodInvocationMessage, isStreamCall, cts, streamOwner); } if (isStreamCall || isStreamResponse) @@ -559,21 +559,21 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, { wasSemaphoreReleased = !hubCallerClients.TrySetSemaphoreReleased(); } - await CleanupInvocation(connection, streamRegistrations, hubActivator, hub, scope); + await CleanupInvocation(connection, hubMethodInvocationMessage, streamOwner, hubActivator, hub, scope); } } return !wasSemaphoreReleased; } - private static ValueTask CleanupInvocation(HubConnectionContext connection, List? streamRegistrations, IHubActivator? hubActivator, + private static ValueTask CleanupInvocation(HubConnectionContext connection, HubMethodInvocationMessage hubMessage, object? streamOwner, IHubActivator? hubActivator, THub? hub, AsyncServiceScope scope) { - if (streamRegistrations is not null) + if (streamOwner is not null) { - foreach (var streamRegistration in streamRegistrations) + foreach (var streamId in hubMessage.StreamIds!) { - connection.StreamTracker.TryComplete(streamRegistration); + connection.StreamTracker.TryComplete(streamId, streamOwner); } } @@ -587,7 +587,7 @@ private static ValueTask CleanupInvocation(HubConnectionContext connection, List private async Task StreamAsync(string invocationId, HubConnectionContext connection, HubCallerContext hubCallerContext, object?[] arguments, AsyncServiceScope scope, IHubActivator hubActivator, THub hub, CancellationTokenSource? streamCts, HubMethodInvocationMessage hubMethodInvocationMessage, - HubMethodDescriptor descriptor, List? streamRegistrations) + HubMethodDescriptor descriptor, object? streamOwner) { string? error = null; @@ -674,7 +674,7 @@ private async Task StreamAsync(string invocationId, HubConnectionContext connect Activity.Current = previousActivity; } - await CleanupInvocation(connection, streamRegistrations, hubActivator, hub, scope); + await CleanupInvocation(connection, hubMethodInvocationMessage, streamOwner, hubActivator, hub, scope); // Only remove/dispose the CTS if we successfully registered it, otherwise we'd evict // another invocation's CTS on ID collision. @@ -821,7 +821,7 @@ await connection.WriteAsync(CompletionMessage.WithError(hubMethodInvocationMessa private void ReplaceArguments(HubMethodDescriptor descriptor, HubMethodInvocationMessage hubMethodInvocationMessage, bool isStreamCall, HubConnectionContext connection, AsyncServiceScope scope, ref object?[] arguments, - ref List? streamRegistrations, out CancellationTokenSource? cts) + ref object? streamOwner, out CancellationTokenSource? cts) { cts = null; // In order to add the synthetic arguments we need a new array because the invocation array is too small (it doesn't know about synthetic arguments) @@ -846,8 +846,7 @@ private void ReplaceArguments(HubMethodDescriptor descriptor, HubMethodInvocatio Log.StartingParameterStream(_logger, hubMethodInvocationMessage.StreamIds![streamPointer]); var itemType = descriptor.StreamingParameters![streamPointer]; arguments[parameterPointer] = connection.StreamTracker.AddStream(hubMethodInvocationMessage.StreamIds[streamPointer], - itemType, descriptor.OriginalParameterTypes[parameterPointer], out var streamRegistration); - (streamRegistrations ??= []).Add(streamRegistration); + itemType, descriptor.OriginalParameterTypes[parameterPointer], streamOwner ??= new object()); streamPointer++; } diff --git a/src/SignalR/server/Core/src/StreamTracker.cs b/src/SignalR/server/Core/src/StreamTracker.cs index 156628a6778c..e812be3ec534 100644 --- a/src/SignalR/server/Core/src/StreamTracker.cs +++ b/src/SignalR/server/Core/src/StreamTracker.cs @@ -16,7 +16,7 @@ internal sealed class StreamTracker { private static readonly MethodInfo _buildConverterMethod = typeof(StreamTracker).GetMethods(BindingFlags.NonPublic | BindingFlags.Static).Single(m => m.Name.Equals(nameof(BuildStream))); private readonly object[] _streamConverterArgs; - private readonly ConcurrentDictionary _lookup = new ConcurrentDictionary(); + private readonly ConcurrentDictionary _lookup = new ConcurrentDictionary(); public StreamTracker(int streamBufferCapacity) { @@ -24,30 +24,29 @@ public StreamTracker(int streamBufferCapacity) } /// - /// Creates a new stream and returns the ChannelReader for it as an object and a registration used for cleanup. + /// Creates a new stream and returns the ChannelReader for it as an object. /// [UnconditionalSuppressMessage("Trimming", "IL2060:MakeGenericMethod", Justification = "BuildStream doesn't have trimming annotations.")] [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "HubMethodDescriptor checks for ValueType streaming item types when PublishAot=true. Developers will get an exception in this situation before publishing.")] - public object AddStream(string streamId, Type itemType, Type targetType, out StreamRegistration streamRegistration) + public object AddStream(string streamId, Type itemType, Type targetType, object streamOwner) { Debug.Assert(RuntimeFeature.IsDynamicCodeSupported || !itemType.IsValueType, "HubMethodDescriptor ensures itemType is not a ValueType when PublishAot=true."); var newConverter = (IStreamConverter)_buildConverterMethod.MakeGenericMethod(itemType).Invoke(null, _streamConverterArgs)!; var reader = newConverter.GetReaderAsObject(targetType); - if (!_lookup.TryAdd(streamId, newConverter)) + if (!_lookup.TryAdd(streamId, new StreamRegistration(streamOwner, newConverter))) { throw new HubException($"Stream ID '{streamId}' is already in use."); } - streamRegistration = new StreamRegistration(streamId, newConverter); return reader; } - private bool TryGetConverter(string streamId, [NotNullWhen(true)] out IStreamConverter? converter) + private bool TryGetRegistration(string streamId, [NotNullWhen(true)] out StreamRegistration? registration) { - if (_lookup.TryGetValue(streamId, out converter)) + if (_lookup.TryGetValue(streamId, out registration)) { return true; } @@ -57,9 +56,9 @@ private bool TryGetConverter(string streamId, [NotNullWhen(true)] out IStreamCon public bool TryProcessItem(StreamItemMessage message, [NotNullWhen(true)] out Task? task) { - if (TryGetConverter(message.InvocationId!, out var converter)) + if (TryGetRegistration(message.InvocationId!, out var registration)) { - task = converter.WriteToStream(message.Item); + task = registration.Converter.WriteToStream(message.Item); return true; } @@ -69,9 +68,9 @@ public bool TryProcessItem(StreamItemMessage message, [NotNullWhen(true)] out Ta public Type GetStreamItemType(string streamId) { - if (TryGetConverter(streamId, out var converter)) + if (TryGetRegistration(streamId, out var registration)) { - return converter.GetItemType(); + return registration.Converter.GetItemType(); } throw new KeyNotFoundException($"No stream with id '{streamId}' could be found."); @@ -79,21 +78,22 @@ public Type GetStreamItemType(string streamId) public bool TryComplete(CompletionMessage message) { - _lookup.TryRemove(message.InvocationId!, out var converter); - if (converter is null) + _lookup.TryRemove(message.InvocationId!, out var registration); + if (registration is null) { return false; } - converter.TryComplete(message.HasResult || message.Error == null ? null : new HubException(message.Error)); + registration.Converter.TryComplete(message.HasResult || message.Error == null ? null : new HubException(message.Error)); return true; } - public bool TryComplete(StreamRegistration streamRegistration) + public bool TryComplete(string streamId, object streamOwner) { - var stream = new KeyValuePair(streamRegistration.StreamId, streamRegistration.Converter); - if (((ICollection>)_lookup).Remove(stream)) + if (_lookup.TryGetValue(streamId, out var registration) && + ReferenceEquals(registration.Owner, streamOwner) && + ((ICollection>)_lookup).Remove(new KeyValuePair(streamId, registration))) { - streamRegistration.Converter.TryComplete(null); + registration.Converter.TryComplete(null); return true; } @@ -104,7 +104,7 @@ public void CompleteAll(Exception ex) { foreach (var converter in _lookup) { - converter.Value.TryComplete(ex); + converter.Value.Converter.TryComplete(ex); } } @@ -113,20 +113,20 @@ private static IStreamConverter BuildStream(int streamBufferCapacity) return new ChannelConverter(streamBufferCapacity); } - public readonly struct StreamRegistration + private sealed class StreamRegistration { - internal StreamRegistration(string streamId, IStreamConverter converter) + public StreamRegistration(object owner, IStreamConverter converter) { - StreamId = streamId; + Owner = owner; Converter = converter; } - internal string StreamId { get; } + public object Owner { get; } - internal IStreamConverter Converter { get; } + public IStreamConverter Converter { get; } } - internal interface IStreamConverter + private interface IStreamConverter { Type GetItemType(); object GetReaderAsObject(Type type); From e96b45584bf5558fd9444a72f3d9b1e0206e2b49 Mon Sep 17 00:00:00 2001 From: Javier Calvarro Nelson <6995051+javiercn@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:29:15 +0200 Subject: [PATCH 3/7] Avoid upload stream ownership allocations Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../Core/src/Internal/DefaultHubDispatcher.cs | 14 +++++------ src/SignalR/server/Core/src/StreamTracker.cs | 23 +++++++++++-------- 2 files changed, 21 insertions(+), 16 deletions(-) diff --git a/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs b/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs index 3224ab4abf0f..ad7e460cffb0 100644 --- a/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs +++ b/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs @@ -370,7 +370,7 @@ private async Task Invoke(HubMethodDescriptor descriptor, HubConnectionCon var scope = _serviceScopeFactory.CreateAsyncScope(); IHubActivator? hubActivator = null; THub? hub = null; - object? streamOwner = null; + long? streamOwner = null; try { hubActivator = scope.ServiceProvider.GetRequiredService>(); @@ -441,7 +441,7 @@ static async Task ExecuteInvocation(DefaultHubDispatcher dispatcher, HubMethodInvocationMessage hubMethodInvocationMessage, bool isStreamCall, CancellationTokenSource? cts, - object? streamOwner) + long? streamOwner) { var logger = dispatcher._logger; var enableDetailedErrors = dispatcher._enableDetailedErrors; @@ -566,14 +566,14 @@ await SendInvocationError(hubMethodInvocationMessage.InvocationId, connection, return !wasSemaphoreReleased; } - private static ValueTask CleanupInvocation(HubConnectionContext connection, HubMethodInvocationMessage hubMessage, object? streamOwner, IHubActivator? hubActivator, + private static ValueTask CleanupInvocation(HubConnectionContext connection, HubMethodInvocationMessage hubMessage, long? streamOwner, IHubActivator? hubActivator, THub? hub, AsyncServiceScope scope) { if (streamOwner is not null) { foreach (var streamId in hubMessage.StreamIds!) { - connection.StreamTracker.TryComplete(streamId, streamOwner); + connection.StreamTracker.TryComplete(streamId, streamOwner.Value); } } @@ -587,7 +587,7 @@ private static ValueTask CleanupInvocation(HubConnectionContext connection, HubM private async Task StreamAsync(string invocationId, HubConnectionContext connection, HubCallerContext hubCallerContext, object?[] arguments, AsyncServiceScope scope, IHubActivator hubActivator, THub hub, CancellationTokenSource? streamCts, HubMethodInvocationMessage hubMethodInvocationMessage, - HubMethodDescriptor descriptor, object? streamOwner) + HubMethodDescriptor descriptor, long? streamOwner) { string? error = null; @@ -821,7 +821,7 @@ await connection.WriteAsync(CompletionMessage.WithError(hubMethodInvocationMessa private void ReplaceArguments(HubMethodDescriptor descriptor, HubMethodInvocationMessage hubMethodInvocationMessage, bool isStreamCall, HubConnectionContext connection, AsyncServiceScope scope, ref object?[] arguments, - ref object? streamOwner, out CancellationTokenSource? cts) + ref long? streamOwner, out CancellationTokenSource? cts) { cts = null; // In order to add the synthetic arguments we need a new array because the invocation array is too small (it doesn't know about synthetic arguments) @@ -846,7 +846,7 @@ private void ReplaceArguments(HubMethodDescriptor descriptor, HubMethodInvocatio Log.StartingParameterStream(_logger, hubMethodInvocationMessage.StreamIds![streamPointer]); var itemType = descriptor.StreamingParameters![streamPointer]; arguments[parameterPointer] = connection.StreamTracker.AddStream(hubMethodInvocationMessage.StreamIds[streamPointer], - itemType, descriptor.OriginalParameterTypes[parameterPointer], streamOwner ??= new object()); + itemType, descriptor.OriginalParameterTypes[parameterPointer], streamOwner ??= connection.StreamTracker.GetNextStreamOwner()); streamPointer++; } diff --git a/src/SignalR/server/Core/src/StreamTracker.cs b/src/SignalR/server/Core/src/StreamTracker.cs index e812be3ec534..4fa3b85837fb 100644 --- a/src/SignalR/server/Core/src/StreamTracker.cs +++ b/src/SignalR/server/Core/src/StreamTracker.cs @@ -17,12 +17,18 @@ internal sealed class StreamTracker private static readonly MethodInfo _buildConverterMethod = typeof(StreamTracker).GetMethods(BindingFlags.NonPublic | BindingFlags.Static).Single(m => m.Name.Equals(nameof(BuildStream))); private readonly object[] _streamConverterArgs; private readonly ConcurrentDictionary _lookup = new ConcurrentDictionary(); + private long _nextStreamOwner; public StreamTracker(int streamBufferCapacity) { _streamConverterArgs = new object[] { streamBufferCapacity }; } + public long GetNextStreamOwner() + { + return Interlocked.Increment(ref _nextStreamOwner); + } + /// /// Creates a new stream and returns the ChannelReader for it as an object. /// @@ -30,7 +36,7 @@ public StreamTracker(int streamBufferCapacity) Justification = "BuildStream doesn't have trimming annotations.")] [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "HubMethodDescriptor checks for ValueType streaming item types when PublishAot=true. Developers will get an exception in this situation before publishing.")] - public object AddStream(string streamId, Type itemType, Type targetType, object streamOwner) + public object AddStream(string streamId, Type itemType, Type targetType, long streamOwner) { Debug.Assert(RuntimeFeature.IsDynamicCodeSupported || !itemType.IsValueType, "HubMethodDescriptor ensures itemType is not a ValueType when PublishAot=true."); @@ -44,7 +50,7 @@ public object AddStream(string streamId, Type itemType, Type targetType, object return reader; } - private bool TryGetRegistration(string streamId, [NotNullWhen(true)] out StreamRegistration? registration) + private bool TryGetRegistration(string streamId, out StreamRegistration registration) { if (_lookup.TryGetValue(streamId, out registration)) { @@ -78,8 +84,7 @@ public Type GetStreamItemType(string streamId) public bool TryComplete(CompletionMessage message) { - _lookup.TryRemove(message.InvocationId!, out var registration); - if (registration is null) + if (!_lookup.TryRemove(message.InvocationId!, out var registration)) { return false; } @@ -87,10 +92,10 @@ public bool TryComplete(CompletionMessage message) return true; } - public bool TryComplete(string streamId, object streamOwner) + public bool TryComplete(string streamId, long streamOwner) { if (_lookup.TryGetValue(streamId, out var registration) && - ReferenceEquals(registration.Owner, streamOwner) && + registration.Owner == streamOwner && ((ICollection>)_lookup).Remove(new KeyValuePair(streamId, registration))) { registration.Converter.TryComplete(null); @@ -113,15 +118,15 @@ private static IStreamConverter BuildStream(int streamBufferCapacity) return new ChannelConverter(streamBufferCapacity); } - private sealed class StreamRegistration + private readonly struct StreamRegistration { - public StreamRegistration(object owner, IStreamConverter converter) + public StreamRegistration(long owner, IStreamConverter converter) { Owner = owner; Converter = converter; } - public object Owner { get; } + public long Owner { get; } public IStreamConverter Converter { get; } } From 1280ed7e104cb988abfce3a860c9a574f9e57f67 Mon Sep 17 00:00:00 2001 From: Javier Calvarro Nelson <6995051+javiercn@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:29:15 +0200 Subject: [PATCH 4/7] Simplify upload stream registration ownership Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/SignalR/server/Core/src/StreamTracker.cs | 34 +++++++------------- 1 file changed, 12 insertions(+), 22 deletions(-) diff --git a/src/SignalR/server/Core/src/StreamTracker.cs b/src/SignalR/server/Core/src/StreamTracker.cs index 4fa3b85837fb..836db6f704bf 100644 --- a/src/SignalR/server/Core/src/StreamTracker.cs +++ b/src/SignalR/server/Core/src/StreamTracker.cs @@ -16,7 +16,7 @@ internal sealed class StreamTracker { private static readonly MethodInfo _buildConverterMethod = typeof(StreamTracker).GetMethods(BindingFlags.NonPublic | BindingFlags.Static).Single(m => m.Name.Equals(nameof(BuildStream))); private readonly object[] _streamConverterArgs; - private readonly ConcurrentDictionary _lookup = new ConcurrentDictionary(); + private readonly ConcurrentDictionary _lookup = new(); private long _nextStreamOwner; public StreamTracker(int streamBufferCapacity) @@ -42,7 +42,7 @@ public object AddStream(string streamId, Type itemType, Type targetType, long st var newConverter = (IStreamConverter)_buildConverterMethod.MakeGenericMethod(itemType).Invoke(null, _streamConverterArgs)!; var reader = newConverter.GetReaderAsObject(targetType); - if (!_lookup.TryAdd(streamId, new StreamRegistration(streamOwner, newConverter))) + if (!_lookup.TryAdd(streamId, (streamOwner, newConverter))) { throw new HubException($"Stream ID '{streamId}' is already in use."); } @@ -50,7 +50,7 @@ public object AddStream(string streamId, Type itemType, Type targetType, long st return reader; } - private bool TryGetRegistration(string streamId, out StreamRegistration registration) + private bool TryGetRegistration(string streamId, out (long Owner, IStreamConverter Converter) registration) { if (_lookup.TryGetValue(streamId, out registration)) { @@ -94,15 +94,18 @@ public bool TryComplete(CompletionMessage message) public bool TryComplete(string streamId, long streamOwner) { - if (_lookup.TryGetValue(streamId, out var registration) && - registration.Owner == streamOwner && - ((ICollection>)_lookup).Remove(new KeyValuePair(streamId, registration))) + if (!_lookup.TryGetValue(streamId, out var registration) || registration.Owner != streamOwner) { - registration.Converter.TryComplete(null); - return true; + return false; } - return false; + if (!_lookup.TryRemove(KeyValuePair.Create(streamId, registration))) + { + return false; + } + + registration.Converter.TryComplete(null); + return true; } public void CompleteAll(Exception ex) @@ -118,19 +121,6 @@ private static IStreamConverter BuildStream(int streamBufferCapacity) return new ChannelConverter(streamBufferCapacity); } - private readonly struct StreamRegistration - { - public StreamRegistration(long owner, IStreamConverter converter) - { - Owner = owner; - Converter = converter; - } - - public long Owner { get; } - - public IStreamConverter Converter { get; } - } - private interface IStreamConverter { Type GetItemType(); From f866feeb5672e91647d8c435632c11eb60c354b5 Mon Sep 17 00:00:00 2001 From: Javier Calvarro Nelson <6995051+javiercn@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:29:16 +0200 Subject: [PATCH 5/7] Defer upload stream reader creation Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 82e97f5a-a052-4dbe-9cf1-b62f45cf7ee2 --- src/SignalR/server/Core/src/StreamTracker.cs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/SignalR/server/Core/src/StreamTracker.cs b/src/SignalR/server/Core/src/StreamTracker.cs index 836db6f704bf..7666cd420b06 100644 --- a/src/SignalR/server/Core/src/StreamTracker.cs +++ b/src/SignalR/server/Core/src/StreamTracker.cs @@ -41,13 +41,12 @@ public object AddStream(string streamId, Type itemType, Type targetType, long st Debug.Assert(RuntimeFeature.IsDynamicCodeSupported || !itemType.IsValueType, "HubMethodDescriptor ensures itemType is not a ValueType when PublishAot=true."); var newConverter = (IStreamConverter)_buildConverterMethod.MakeGenericMethod(itemType).Invoke(null, _streamConverterArgs)!; - var reader = newConverter.GetReaderAsObject(targetType); if (!_lookup.TryAdd(streamId, (streamOwner, newConverter))) { throw new HubException($"Stream ID '{streamId}' is already in use."); } - return reader; + return newConverter.GetReaderAsObject(targetType); } private bool TryGetRegistration(string streamId, out (long Owner, IStreamConverter Converter) registration) From 8d18294f91eb7656fee0286d6e826251ac00b2fe Mon Sep 17 00:00:00 2001 From: Javier Calvarro Nelson <6995051+javiercn@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:29:16 +0200 Subject: [PATCH 6/7] Dispose cancellation source after binding failure Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 82e97f5a-a052-4dbe-9cf1-b62f45cf7ee2 --- .../Core/src/Internal/DefaultHubDispatcher.cs | 69 +++++++++++-------- 1 file changed, 41 insertions(+), 28 deletions(-) diff --git a/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs b/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs index ad7e460cffb0..279fd103e2e4 100644 --- a/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs +++ b/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs @@ -829,39 +829,52 @@ private void ReplaceArguments(HubMethodDescriptor descriptor, HubMethodInvocatio var streamPointer = 0; var hubInvocationArgumentPointer = 0; - for (var parameterPointer = 0; parameterPointer < arguments.Length; parameterPointer++) + var argumentsReplaced = false; + try { - // populate the synthetic arguments first - if (descriptor.IsServiceArgument(parameterPointer)) - { - arguments[parameterPointer] = descriptor.GetService(scope.ServiceProvider, parameterPointer, descriptor.OriginalParameterTypes[parameterPointer]); - } - else if (descriptor.OriginalParameterTypes[parameterPointer] == typeof(CancellationToken)) + for (var parameterPointer = 0; parameterPointer < arguments.Length; parameterPointer++) { - cts = CancellationTokenSource.CreateLinkedTokenSource(connection.ConnectionAborted); - arguments[parameterPointer] = cts.Token; - } - else if (isStreamCall && ReflectionHelper.IsStreamingType(descriptor.OriginalParameterTypes[parameterPointer], mustBeDirectType: true)) - { - Log.StartingParameterStream(_logger, hubMethodInvocationMessage.StreamIds![streamPointer]); - var itemType = descriptor.StreamingParameters![streamPointer]; - arguments[parameterPointer] = connection.StreamTracker.AddStream(hubMethodInvocationMessage.StreamIds[streamPointer], - itemType, descriptor.OriginalParameterTypes[parameterPointer], streamOwner ??= connection.StreamTracker.GetNextStreamOwner()); + // populate the synthetic arguments first + if (descriptor.IsServiceArgument(parameterPointer)) + { + arguments[parameterPointer] = descriptor.GetService(scope.ServiceProvider, parameterPointer, descriptor.OriginalParameterTypes[parameterPointer]); + } + else if (descriptor.OriginalParameterTypes[parameterPointer] == typeof(CancellationToken)) + { + cts = CancellationTokenSource.CreateLinkedTokenSource(connection.ConnectionAborted); + arguments[parameterPointer] = cts.Token; + } + else if (isStreamCall && ReflectionHelper.IsStreamingType(descriptor.OriginalParameterTypes[parameterPointer], mustBeDirectType: true)) + { + Log.StartingParameterStream(_logger, hubMethodInvocationMessage.StreamIds![streamPointer]); + var itemType = descriptor.StreamingParameters![streamPointer]; + arguments[parameterPointer] = connection.StreamTracker.AddStream(hubMethodInvocationMessage.StreamIds[streamPointer], + itemType, descriptor.OriginalParameterTypes[parameterPointer], streamOwner ??= connection.StreamTracker.GetNextStreamOwner()); - streamPointer++; - } - else if (hubMethodInvocationMessage.Arguments?.Length > hubInvocationArgumentPointer && - (hubMethodInvocationMessage.Arguments[hubInvocationArgumentPointer] == null || - descriptor.OriginalParameterTypes[parameterPointer].IsAssignableFrom(hubMethodInvocationMessage.Arguments[hubInvocationArgumentPointer]?.GetType()))) - { - // The types match so it isn't a synthetic argument, just copy it into the arguments array - arguments[parameterPointer] = hubMethodInvocationMessage.Arguments[hubInvocationArgumentPointer]; - hubInvocationArgumentPointer++; + streamPointer++; + } + else if (hubMethodInvocationMessage.Arguments?.Length > hubInvocationArgumentPointer && + (hubMethodInvocationMessage.Arguments[hubInvocationArgumentPointer] == null || + descriptor.OriginalParameterTypes[parameterPointer].IsAssignableFrom(hubMethodInvocationMessage.Arguments[hubInvocationArgumentPointer]?.GetType()))) + { + // The types match so it isn't a synthetic argument, just copy it into the arguments array + arguments[parameterPointer] = hubMethodInvocationMessage.Arguments[hubInvocationArgumentPointer]; + hubInvocationArgumentPointer++; + } + else + { + // This should never happen + Debug.Assert(false, $"Failed to bind argument of type '{descriptor.OriginalParameterTypes[parameterPointer].Name}' for hub method '{descriptor.MethodExecutor.MethodInfo.Name}'."); + } } - else + + argumentsReplaced = true; + } + finally + { + if (!argumentsReplaced) { - // This should never happen - Debug.Assert(false, $"Failed to bind argument of type '{descriptor.OriginalParameterTypes[parameterPointer].Name}' for hub method '{descriptor.MethodExecutor.MethodInfo.Name}'."); + cts?.Dispose(); } } } From 41b010c6752186a49ee1494db262796c52e5f679 Mon Sep 17 00:00:00 2001 From: Javier Calvarro Nelson <6995051+javiercn@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:29:17 +0200 Subject: [PATCH 7/7] Reuse upload stream test helper Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 82e97f5a-a052-4dbe-9cf1-b62f45cf7ee2 --- .../HubConnectionHandlerTestUtils/Hubs.cs | 16 +--------------- 1 file changed, 1 insertion(+), 15 deletions(-) diff --git a/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTestUtils/Hubs.cs b/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTestUtils/Hubs.cs index 2e8fdbde2371..03328940668d 100644 --- a/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTestUtils/Hubs.cs +++ b/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests/HubConnectionHandlerTestUtils/Hubs.cs @@ -225,21 +225,7 @@ public async Task StreamingConcat(ChannelReader source) public async Task StreamingConcatTwoStreams(ChannelReader first, ChannelReader second) { - return await ReadStream(first) + await ReadStream(second); - - static async Task ReadStream(ChannelReader source) - { - var result = new StringBuilder(); - while (await source.WaitToReadAsync()) - { - while (source.TryRead(out var item)) - { - result.Append(item); - } - } - - return result.ToString(); - } + return await StreamingConcat(first) + await StreamingConcat(second); } public async Task StreamDontRead(ChannelReader source)