[Bug]: Server stuck writing to stream after client disconnect

### Duplicate ?

- [x] I have searched issues/discussions and did not find other issues/discussions reporting this bug.

### Product version

1.7.0

### Describe expected behavior

Background: I tried to transfer an unbounded stream across WCF. I now recognise this isn't intended use of WCF streams and that it was never going to work. However, a few people have also suggested that I may have uncovered a bug in CoreWCF and to raise it here.

To recreate, create an unbounded stream which does not have a length and has no EOF, .e.g:

```csharp
namespace RandomNumberCore;

public class RandomStream : Stream
{
public RandomStream(Random random)
{
this._random = random;
}

private int _sequence;
private readonly Random _random;
public override bool CanRead => true;

public override bool CanSeek => false;

public override bool CanWrite => false;

public override long Length => throw new NotSupportedException();

// ReSharper disable once ValueParameterNotUsed
public override long Position { get => _sequence; set => throw new NotSupportedException(); }

public override void Flush()
{}

public override int Read(byte[] buffer, int offset, int count)
{
var internalBuffer = new Span(buffer, offset, count);
_random.NextBytes(internalBuffer);
_sequence+=count;
return count;
}

public override int Read(Span buffer)
{
_random.NextBytes(buffer);
_sequence+=buffer.Length;
return buffer.Length;
}

public override long Seek(long offset, SeekOrigin origin)
{
throw new NotSupportedException();
}

public override void SetLength(long value)
{
throw new NotSupportedException();
}

public override void Write(byte[] buffer, int offset, int count)
{
throw new NotSupportedException();
}
}
```

```csharp
namespace RandomNumberCore;

[ServiceContract]
public interface IStreamingService
{
[OperationContract]
Stream GetRandomStream();
}

public class StreamingService : IStreamingService
{
public Stream GetRandomStream()
{
return new RandomStream(Random.Shared);
}
}
```

Host this in Kestrel:

```csharp
using System.Diagnostics;

var builder = WebApplication.CreateBuilder();

builder.Services.AddServiceModelServices();
builder.Services.AddServiceModelMetadata();
builder.Services.AddSingleton();
builder.WebHost.UseKestrel(options =>
{
options.AllowSynchronousIO = true;
options.ListenLocalhost(7151, listenOptions =>
{
listenOptions.UseHttps();
if (Debugger.IsAttached)
{
listenOptions.UseConnectionLogging();
}
});
});

var app = builder.Build();

app.UseServiceModel(serviceBuilder =>
{
serviceBuilder.AddService();
serviceBuilder.AddServiceEndpoint(new BasicHttpBinding(BasicHttpSecurityMode.Transport){TransferMode = TransferMode.Streamed}, $"https://localhost:7151/StreamingService.svc");

var serviceMetadataBehavior = app.Services.GetRequiredService();
serviceMetadataBehavior.HttpsGetEnabled = true;
});

app.Run();
```

Request this via WCF.

Full recreation code is at https://github.com/richardcocks/randomNumberStackOverflow .

### Describe actual behavior

After the client retrieves bytes from the stream and disconnects, the server is left writing indefinitely to the stream.

With mutiple requests, the server can quickly become overwhelmed by the processing required to generate and write bytes.

It also takes a long time to then stop the service.

### Which binding

BasicHttp

### security

Transport

### Which .NET version

.NET 8

### Which os platform

Windows

### Code snippet used to reproduce the issue

```c#

```

### Stacktrace if any

```shell
Thread (0x4BD4):
CPU_TIME
Microsoft.AspNetCore.Server.Kestrel.Core.il!Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.Http1OutputProducer.WriteChunkAsync(value class System.ReadOnlySpan`1,value class System.Threading.CancellationToken)
Microsoft.AspNetCore.Server.Kestrel.Core.il!Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpProtocol.WritePipeAsync(value class System.ReadOnlyMemory`1,value class System.Threading.CancellationToken)
System.Private.DataContractSerialization.il!System.Xml.XmlBaseWriter.WriteBase64(unsigned int8[],int32,int32)
System.Private.DataContractSerialization.il!System.Xml.XmlDictionaryWriter.WriteValue(class System.Xml.IStreamProvider)
System.Private.DataContractSerialization.il!System.Xml.XmlDictionaryWriter.WriteValueAsync(class System.Xml.IStreamProvider)
CoreWCF.Primitives!CoreWCF.Dispatcher.StreamFormatter+d__7.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Primitives!CoreWCF.Dispatcher.StreamFormatter.SerializeAsync(class System.Xml.XmlDictionaryWriter,class System.Object[],class System.Object)
CoreWCF.Primitives!CoreWCF.Dispatcher.OperationFormatter+d__34.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Primitives!CoreWCF.Dispatcher.OperationFormatter.SerializeBodyContentsAsync(class System.Xml.XmlDictionaryWriter,class CoreWCF.Channels.MessageVersion,class System.Object[],class System.Object,bool)
CoreWCF.Primitives!CoreWCF.Dispatcher.OperationFormatter+OperationFormatterMessage+OperationFormatterBodyWriter.OnWriteBodyContentsAsync(class System.Xml.XmlDictionaryWriter)
CoreWCF.Primitives!CoreWCF.Channels.BodyWriter.WriteBodyContentsAsync(class System.Xml.XmlDictionaryWriter)
CoreWCF.Primitives!CoreWCF.Channels.BodyWriterMessage.OnWriteBodyContentsAsync(class System.Xml.XmlDictionaryWriter)
CoreWCF.Primitives!CoreWCF.Channels.BodyWriterMessage+d__19.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Primitives!CoreWCF.Channels.BodyWriterMessage.OnWriteMessageAsync(class System.Xml.XmlDictionaryWriter)
CoreWCF.Primitives!CoreWCF.Channels.Message+d__73.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Primitives!CoreWCF.Channels.Message.WriteMessageAsync(class System.Xml.XmlDictionaryWriter)
CoreWCF.Primitives!CoreWCF.Channels.TextMessageEncoderFactory+TextMessageEncoder+d__39.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Primitives!CoreWCF.Channels.TextMessageEncoderFactory+TextMessageEncoder.WriteMessageAsync(class CoreWCF.Channels.Message,class System.IO.Stream)
CoreWCF.Http!CoreWCF.Channels.HttpOutput+d__43.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Http!CoreWCF.Channels.HttpOutput.WriteStreamedMessageAsync(value class System.Threading.CancellationToken)
CoreWCF.Http!CoreWCF.Channels.HttpOutput+d__45.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Http!CoreWCF.Channels.HttpOutput.SendAsync(value class System.Threading.CancellationToken)
CoreWCF.Http!CoreWCF.Channels.HttpRequestContext+d__25.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Http!CoreWCF.Channels.HttpRequestContext.OnReplyAsync(class CoreWCF.Channels.Message,value class System.Threading.CancellationToken)
CoreWCF.Http!CoreWCF.Channels.RequestContextBase+d__37.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Http!CoreWCF.Channels.RequestContextBase.ReplyAsync(class CoreWCF.Channels.Message,value class System.Threading.CancellationToken)
CoreWCF.Primitives!CoreWCF.Dispatcher.ImmutableDispatchRuntime+d__53.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder`1[System.__Canon].Start(!!0&)
CoreWCF.Primitives!CoreWCF.Dispatcher.ImmutableDispatchRuntime.ReplyAsync(class CoreWCF.Dispatcher.MessageRpc)
CoreWCF.Primitives!CoreWCF.Dispatcher.ImmutableDispatchRuntime+d__66.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start(!!0&)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder.Start(!!0&)
CoreWCF.Primitives!CoreWCF.Dispatcher.ImmutableDispatchRuntime.ProcessError(class CoreWCF.Dispatcher.MessageRpc)
CoreWCF.Primitives!CoreWCF.Dispatcher.ImmutableDispatchRuntime+d__65.MoveNext()
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder`1+AsyncStateMachineBox`1[System.__Canon,CoreWCF.Dispatcher.ImmutableDispatchRuntime+d__65].ExecutionContextCallback(class System.Object)
System.Private.CoreLib.il!System.Threading.ExecutionContext.RunFromThreadPoolDispatchLoop(class System.Threading.Thread,class System.Threading.ExecutionContext,class System.Threading.ContextCallback,class System.Object)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder`1+AsyncStateMachineBox`1[System.__Canon,CoreWCF.Dispatcher.ImmutableDispatchRuntime+d__65].MoveNext(class System.Threading.Thread)
System.Private.CoreLib.il!System.Runtime.CompilerServices.AsyncTaskMethodBuilder`1+AsyncStateMachineBox`1[System.__Canon,CoreWCF.Dispatcher.ImmutableDispatchRuntime+d__65].ExecuteFromThreadPool(class System.Threading.Thread)
System.Private.CoreLib.il!System.Threading.ThreadPoolWorkQueue.Dispatch()
System.Private.CoreLib.il!System.Threading.PortableThreadPool+WorkerThread.WorkerThreadStart()
```