DbResultStream.cs

public class DbResultStream : LifetimedStream
{
    public DbResultStream(Stream originalStream, IDbConnection connection, IDataReader dataReader) : base(originalStream)
    {
        AddDisposable(dataReader);
        AddDisposable(connection);
    }    
}

public class LifetimedStream : StreamWrapper
{
    private readonly Stack<IDisposable> _disposables = new();
    public bool Disposed { get; private set; }
    
    public LifetimedStream(Stream originalStream) : base(originalStream)
    {
        Disposed = false;
    }
    
    public LifetimedStream AddDisposable(IDisposable disposable)
    {
        _disposables.Push(disposable ?? throw new ArgumentNullException(nameof(disposable)));
        return this;
    }
    
    protected override void Dispose(bool disposing)
    {
        if (Disposed)
        {
            throw new InvalidOperationException("Already disposed");
        }
        base.Dispose(disposing);
        if (disposing)
        {
            while(_disposables.TryPop(out var disposable))
            {
                disposable.Dispose();
            }
        }
        Disposed = true;
    }
    
    public override async ValueTask DisposeAsync()
    {
        if (Disposed)
        {
            throw new InvalidOperationException("Already disposed");
        }
        await base.DisposeAsync();
        while(_disposables.TryPop(out var disposable))
        {
            if (disposable is IAsyncDisposable asyncDisposable)
            {
                await asyncDisposable.DisposeAsync();
            }
            else
            {
                disposable.Dispose();
            }
        }
        Disposed = true;
    }
}

public abstract class StreamWrapper : Stream
{
    private readonly Stream OriginalStream;
    private bool _streamDisposed = false;

    protected StreamWrapper(Stream originalStream)
    {
        OriginalStream = originalStream ?? throw new ArgumentNullException(nameof(originalStream));
    }
    
    public override void Close()
    {
        OriginalStream.Close();
    }

    protected override void Dispose(bool disposing)
    {
        if (!disposing || _streamDisposed) return;
        OriginalStream.Dispose();
    }
    public override ValueTask DisposeAsync()
    {
        if (_streamDisposed) return ValueTask.CompletedTask;
        _streamDisposed = true;
        return OriginalStream.DisposeAsync();
    }

    public override bool CanRead => OriginalStream.CanRead;
    public override bool CanSeek => OriginalStream.CanSeek;
    public override bool CanWrite => OriginalStream.CanWrite;
    public override long Length => OriginalStream.Length;
    public override long Position { get => OriginalStream.Position; set => OriginalStream.Position = value; }
    public override void Flush() => OriginalStream.Flush();
    public override int Read(byte[] buffer, int offset, int count) => OriginalStream.Read(buffer, offset, count);
    public override long Seek(long offset, SeekOrigin origin) => OriginalStream.Seek(offset, origin);
    public override void SetLength(long value) => OriginalStream.SetLength(value);
    public override void Write(byte[] buffer, int offset, int count) => OriginalStream.Write(buffer, offset, count);
    public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state) => OriginalStream.BeginRead(buffer, offset, count, callback, state);
    public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state) => OriginalStream.BeginWrite(buffer, offset, count, callback, state);
    public override bool CanTimeout => OriginalStream.CanTimeout;
    public override void CopyTo(Stream destination, int bufferSize) => OriginalStream.CopyTo(destination, bufferSize);
    public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken) => OriginalStream.CopyToAsync(destination, bufferSize, cancellationToken);
    public override int EndRead(IAsyncResult asyncResult) => OriginalStream.EndRead(asyncResult);
    public override void EndWrite(IAsyncResult asyncResult) => OriginalStream.EndWrite(asyncResult);
    public override bool Equals(object? obj) => OriginalStream.Equals(obj);
    public override Task FlushAsync(CancellationToken cancellationToken) => OriginalStream.FlushAsync(cancellationToken);
    public override int GetHashCode() => OriginalStream.GetHashCode();
    public override int Read(Span<byte> buffer) => OriginalStream.Read(buffer);
    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => OriginalStream.ReadAsync(buffer, offset, count, cancellationToken);
    public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) => OriginalStream.ReadAsync(buffer, cancellationToken);
    public override int ReadByte() => OriginalStream.ReadByte();
    public override int ReadTimeout { get => OriginalStream.ReadTimeout; set => OriginalStream.ReadTimeout = value; }
    public override string? ToString() => OriginalStream.ToString();
    public override void Write(ReadOnlySpan<byte> buffer) => OriginalStream.Write(buffer);
    public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => OriginalStream.WriteAsync(buffer, offset, count, cancellationToken);
    public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default) => OriginalStream.WriteAsync(buffer, cancellationToken);
    public override void WriteByte(byte value) => OriginalStream.WriteByte(value);
    public override int WriteTimeout { get => OriginalStream.WriteTimeout; set => OriginalStream.WriteTimeout = value; }
    [Obsolete]
    public override object InitializeLifetimeService() => OriginalStream.InitializeLifetimeService();
}