From 42c3ecd009f4dacde4fd4a678ff0824ced7490eb Mon Sep 17 00:00:00 2001 From: Kenneth Skovhede Date: Fri, 9 May 2025 13:46:45 +0200 Subject: [PATCH] Fix throttle stream issues Some providers, B2, Idrivee2, JottaCloud and S3 (AWS version) needs a checksum for the content. To calculate this content the libraries will calculate the hash by reading the stream. But since the stream can be throttled, the hash calculation is throttled. In some cases the contents are buffered, so the transfer is done at full speed, but the hashing at throttled speed. The average speed is the same, but network traffic is bursty, which it should not be with a throttle. Also added option to disable signing with AWS Sigv4 as that interferes with the throttled stream. It is not currently possible to pass the sha-256 hash to the Sigv4 code, so it will always read the entire (potentially throttled) stream before sending the contents, causing unwanted delays. Use the new option `--s3-disable-payload-signing` to disable v4 signing. --- .../Backend/AliyunOSS/AliyunOSSBackend.cs | 7 +- Duplicati/Library/Backend/Backblaze/B2.cs | 43 +---- .../Backend/Idrivee2/Idrivee2Backend.cs | 3 +- .../Library/Backend/Jottacloud/Jottacloud.cs | 155 +++++++----------- Duplicati/Library/Backend/S3/S3AwsClient.cs | 16 +- Duplicati/Library/Backend/S3/S3Backend.cs | 7 +- .../Library/Utility/OverrideableStream.cs | 5 + Duplicati/Library/Utility/Utility.cs | 89 +++++++++- 8 files changed, 180 insertions(+), 145 deletions(-) diff --git a/Duplicati/Library/Backend/AliyunOSS/AliyunOSSBackend.cs b/Duplicati/Library/Backend/AliyunOSS/AliyunOSSBackend.cs index d77909b41..72b1aaa9e 100644 --- a/Duplicati/Library/Backend/AliyunOSS/AliyunOSSBackend.cs +++ b/Duplicati/Library/Backend/AliyunOSS/AliyunOSSBackend.cs @@ -198,9 +198,14 @@ namespace Duplicati.Library.Backend.AliyunOSS var client = GetClient(); try { + var metadata = new ObjectMetadata + { + ContentLength = stream.Length + }; + using var timeoutStream = stream.ObserveReadTimeout(_timeouts.ReadWriteTimeout, false); var objectResult = await Task.Factory.FromAsync( - (cb, state) => client.BeginPutObject(bucketName, objectName, timeoutStream, cb, state), + (cb, state) => client.BeginPutObject(bucketName, objectName, timeoutStream, metadata, cb, state), client.EndPutObject, null).ConfigureAwait(false); diff --git a/Duplicati/Library/Backend/Backblaze/B2.cs b/Duplicati/Library/Backend/Backblaze/B2.cs index 63c7a04c0..967c41a68 100644 --- a/Duplicati/Library/Backend/Backblaze/B2.cs +++ b/Duplicati/Library/Backend/Backblaze/B2.cs @@ -310,43 +310,8 @@ public class B2 : IStreamingBackend public async Task PutAsync(string remotename, Stream stream, CancellationToken cancelToken) { - TempFile? tmp = null; - - var measure = stream; - while (measure is OverrideableStream os && - os.GetType().GetField("m_basestream", BindingFlags.NonPublic | BindingFlags.Instance)?.GetValue(os) is Stream baseStream) - { - measure = baseStream; - } - - if (measure == null) - throw new Exception($"Unable to unwrap stream from: {stream.GetType()}"); - - string sha1; - if (measure.CanSeek) - { - var p = measure.Position; - - // Compute the hash - using (var hashalg = HashFactory.CreateHasher("SHA1")) - sha1 = Utility.Utility.ByteArrayAsHexString(hashalg.ComputeHash(measure)); - - measure.Position = p; - } - else - { - // No seeking possible, use a temp file - tmp = new TempFile(); - await using (var sr = File.OpenWrite(tmp)) - using (var hasher = HashFactory.CreateHasher("SHA1")) - await using (var hc = new HashCalculatingStream(measure, hasher)) - { - await Utility.Utility.CopyStreamAsync(hc, sr, cancelToken).ConfigureAwait(false); - sha1 = hc.GetFinalHashString(); - } - - stream = File.OpenRead(tmp); - } + (stream, var sha1, var tmp) = await Utility.Utility.CalculateThrottledStreamHash(stream, "SHA1", cancelToken).ConfigureAwait(false); + using var _ = tmp; if (_filecache == null) await RebuildFileCache(cancelToken).ConfigureAwait(false); @@ -400,10 +365,6 @@ public class B2 : IStreamingBackend throw; } - finally - { - tmp?.Dispose(); - } } /// diff --git a/Duplicati/Library/Backend/Idrivee2/Idrivee2Backend.cs b/Duplicati/Library/Backend/Idrivee2/Idrivee2Backend.cs index d4f5d7809..c8c3f5432 100644 --- a/Duplicati/Library/Backend/Idrivee2/Idrivee2Backend.cs +++ b/Duplicati/Library/Backend/Idrivee2/Idrivee2Backend.cs @@ -190,9 +190,8 @@ namespace Duplicati.Library.Backend if (m_s3Client == null) { (var accessKeyId, var accessKeySecret) = m_auth.GetCredentials(); - // TODO: Do not make blocking calls in the constructor var host = await GetRegionEndpointAsync("https://api.idrivee2.com/api/service/get_region_end_point/" + accessKeyId, cancellationToken).ConfigureAwait(false); - m_s3Client = new S3AwsClient(accessKeyId, accessKeySecret, null, host, null, true, false, m_timeouts, m_options); + m_s3Client = new S3AwsClient(accessKeyId, accessKeySecret, null, host, null, true, false, false, m_timeouts, m_options); } return m_s3Client; diff --git a/Duplicati/Library/Backend/Jottacloud/Jottacloud.cs b/Duplicati/Library/Backend/Jottacloud/Jottacloud.cs index ec4e5f30f..805544398 100644 --- a/Duplicati/Library/Backend/Jottacloud/Jottacloud.cs +++ b/Duplicati/Library/Backend/Jottacloud/Jottacloud.cs @@ -489,108 +489,71 @@ namespace Duplicati.Library.Backend // we specify the file size in a custom header. And if the stream is not seek-able we are not able // to use stream.Length, so we are back at square one. - // TODO: This unwrapping code is shared with CloudFiles - TempFile? tmpFile = null; - Stream? baseStream = stream; - while (baseStream is OverrideableStream) - baseStream = typeof(OverrideableStream).GetField("m_basestream", System.Reflection.BindingFlags.DeclaredOnly | System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)?.GetValue(baseStream) as System.IO.Stream; - if (baseStream == null) - throw new Exception(string.Format("Unable to unwrap stream from: {0}", stream.GetType())); - string md5Hash; - if (baseStream.CanSeek) - { - var originalPosition = baseStream.Position; - using (var md5 = MD5.Create()) - md5Hash = Utility.Utility.ByteArrayAsHexString(md5.ComputeHash(baseStream)); - baseStream.Position = originalPosition; - } - else - { - // No seeking possible, use a temp file - tmpFile = new TempFile(); - using (var os = File.OpenWrite(tmpFile)) - using (var hasher = MD5.Create()) - using (var md5 = new HashCalculatingStream(baseStream, hasher)) - { - await Utility.Utility.CopyStreamAsync(md5, os, true, cancelToken).ConfigureAwait(false); - md5Hash = md5.GetFinalHashString(); - } - stream = File.OpenRead(tmpFile); - } - try - { - // Create request, with query parameter, and a few custom headers. - // NB: If we wanted to we could send the same POST request as below but without the file contents - // and with "cphash=[md5Hash]" as the only query parameter. Then we will get an HTTP 200 (OK) response - // if an identical file already exists, and we can skip uploading the new file. We will get - // HTTP 404 (Not Found) if file does not exists or it exists with a different hash, in which - // case we must send a new request to upload the new content. - var fileSize = stream.Length; - var req = CreateRequest(WebRequestMethods.Http.Post, remotename, "umode=nomultipart", true); - req.Headers.Add("JMd5", md5Hash); // Not required, but it will make the server verify the content and mark the file as corrupt if there is a mismatch. - req.Headers.Add("JSize", fileSize.ToString()); // Required, and used to mark file as incomplete if we upload something be the total size of the original file! - // File time stamp headers: Since we are working with a stream here we do not know the local file's timestamps, - // and then we can just omit the JCreated and JModified and let the server automatically set the current time. - //req.Headers.Add("JCreated", timeCreated); - //req.Headers.Add("JModified", timeModified); - req.ContentType = "application/octet-stream"; - req.ContentLength = fileSize; + (stream, var md5Hash, var tmp) = await Utility.Utility.CalculateThrottledStreamHash(stream, "MD5", cancelToken).ConfigureAwait(false); + using var _ = tmp; - // Write post data request - var areq = new AsyncHttpRequest(req); - using (var rs = areq.GetRequestStream()) - using (var ts = rs.ObserveWriteTimeout(m_timeouts.ReadWriteTimeout)) - await Utility.Utility.CopyStreamAsync(stream, rs, true, cancelToken).ConfigureAwait(false); - // Send request, and check response - using (var resp = (HttpWebResponse)areq.GetResponse()) - { - if (resp.StatusCode != HttpStatusCode.Created) - throw new WebException(Strings.Jottacloud.FileUploadError, null, WebExceptionStatus.ProtocolError, resp); + // Create request, with query parameter, and a few custom headers. + // NB: If we wanted to we could send the same POST request as below but without the file contents + // and with "cphash=[md5Hash]" as the only query parameter. Then we will get an HTTP 200 (OK) response + // if an identical file already exists, and we can skip uploading the new file. We will get + // HTTP 404 (Not Found) if file does not exists or it exists with a different hash, in which + // case we must send a new request to upload the new content. + var fileSize = stream.Length; + var req = CreateRequest(WebRequestMethods.Http.Post, remotename, "umode=nomultipart", true); + req.Headers.Add("JMd5", md5Hash); // Not required, but it will make the server verify the content and mark the file as corrupt if there is a mismatch. + req.Headers.Add("JSize", fileSize.ToString()); // Required, and used to mark file as incomplete if we upload something be the total size of the original file! + // File time stamp headers: Since we are working with a stream here we do not know the local file's timestamps, + // and then we can just omit the JCreated and JModified and let the server automatically set the current time. + //req.Headers.Add("JCreated", timeCreated); + //req.Headers.Add("JModified", timeModified); + req.ContentType = "application/octet-stream"; + req.ContentLength = fileSize; - // Request seems to be successful, but we must verify the response XML content to be sure that the file - // was correctly uploaded: The server will verify the JSize header and mark the file as incomplete if - // there was mismatch, and it will verify the JMd5 header and mark the file as corrupt if there was a hash - // mismatch. The returned XML contains a file element, and if upload was error free it contains a single - // child element "currentRevision", which has a "state" child element with the string "COMPLETED". - // If there was a problem we should have a "latestRevision" child element, and this will have state with - // value "INCOMPLETE" or "CORRUPT". If the file was new or had no previous complete versions the latestRevision - // will be the only child, but if not there may also be a "currentRevision" representing the previous - // complete version - and then we need to detect the case where our upload failed but there was an existing - // complete version! - using (var rs = areq.GetResponseStream()) + // Write post data request + var areq = new AsyncHttpRequest(req); + using (var rs = areq.GetRequestStream()) + using (var ts = rs.ObserveWriteTimeout(m_timeouts.ReadWriteTimeout)) + await Utility.Utility.CopyStreamAsync(stream, rs, true, cancelToken).ConfigureAwait(false); + // Send request, and check response + using (var resp = (HttpWebResponse)areq.GetResponse()) + { + if (resp.StatusCode != HttpStatusCode.Created) + throw new WebException(Strings.Jottacloud.FileUploadError, null, WebExceptionStatus.ProtocolError, resp); + + // Request seems to be successful, but we must verify the response XML content to be sure that the file + // was correctly uploaded: The server will verify the JSize header and mark the file as incomplete if + // there was mismatch, and it will verify the JMd5 header and mark the file as corrupt if there was a hash + // mismatch. The returned XML contains a file element, and if upload was error free it contains a single + // child element "currentRevision", which has a "state" child element with the string "COMPLETED". + // If there was a problem we should have a "latestRevision" child element, and this will have state with + // value "INCOMPLETE" or "CORRUPT". If the file was new or had no previous complete versions the latestRevision + // will be the only child, but if not there may also be a "currentRevision" representing the previous + // complete version - and then we need to detect the case where our upload failed but there was an existing + // complete version! + using (var rs = areq.GetResponseStream()) + { + var doc = new System.Xml.XmlDocument(); + try { doc.Load(rs); } + catch (System.Xml.XmlException) { - var doc = new System.Xml.XmlDocument(); - try { doc.Load(rs); } - catch (System.Xml.XmlException) - { - throw new WebException(Strings.Jottacloud.FileUploadError, WebExceptionStatus.ProtocolError); - } - bool uploadCompletedSuccessfully = false; - var xFile = doc["file"]; - if (xFile != null) - { - var xRevState = xFile.SelectSingleNode("latestRevision"); - if (xRevState == null) - { - xRevState = xFile.SelectSingleNode("currentRevision/state"); - if (xRevState != null) - uploadCompletedSuccessfully = xRevState.InnerText == "COMPLETED"; // Success: There is no "latestRevision", only a "currentRevision" (and it specifies the file is complete, but I think it always will). - } - } - if (!uploadCompletedSuccessfully) // Report error (and we just let the incomplete/corrupt file revision stay on the server..) - throw new WebException(Strings.Jottacloud.FileUploadError, WebExceptionStatus.ProtocolError); + throw new WebException(Strings.Jottacloud.FileUploadError, WebExceptionStatus.ProtocolError); } + bool uploadCompletedSuccessfully = false; + var xFile = doc["file"]; + if (xFile != null) + { + var xRevState = xFile.SelectSingleNode("latestRevision"); + if (xRevState == null) + { + xRevState = xFile.SelectSingleNode("currentRevision/state"); + if (xRevState != null) + uploadCompletedSuccessfully = xRevState.InnerText == "COMPLETED"; // Success: There is no "latestRevision", only a "currentRevision" (and it specifies the file is complete, but I think it always will). + } + } + if (!uploadCompletedSuccessfully) // Report error (and we just let the incomplete/corrupt file revision stay on the server..) + throw new WebException(Strings.Jottacloud.FileUploadError, WebExceptionStatus.ProtocolError); } } - finally - { - try - { - if (tmpFile != null) - tmpFile.Dispose(); - } - catch { } - } } } } diff --git a/Duplicati/Library/Backend/S3/S3AwsClient.cs b/Duplicati/Library/Backend/S3/S3AwsClient.cs index d21ea417e..ae2c5957a 100644 --- a/Duplicati/Library/Backend/S3/S3AwsClient.cs +++ b/Duplicati/Library/Backend/S3/S3AwsClient.cs @@ -61,6 +61,10 @@ namespace Duplicati.Library.Backend /// The option to specify if chunk encoding should be used /// private readonly bool m_useChunkEncoding; + /// + /// The option to specify if payload signing should be disabled + /// + private readonly bool m_disablePayloadSigning; /// /// The DNS host of the S3 server @@ -93,7 +97,7 @@ namespace Duplicati.Library.Backend ]); public S3AwsClient(string awsID, string awsKey, string? locationConstraint, string servername, - string? storageClass, bool useSSL, bool disableChunkEncoding, TimeoutOptionsHelper.Timeouts timeouts, Dictionary options) + string? storageClass, bool useSSL, bool disableChunkEncoding, bool disablePayloadSigning, TimeoutOptionsHelper.Timeouts timeouts, Dictionary options) { var cfg = GetDefaultAmazonS3Config(); cfg.UseHttp = !useSSL; @@ -109,6 +113,7 @@ namespace Duplicati.Library.Backend m_storageClass = storageClass; m_dnsHost = string.IsNullOrWhiteSpace(cfg.ServiceURL) ? null : new System.Uri(cfg.ServiceURL).Host; m_useChunkEncoding = !disableChunkEncoding; + m_disablePayloadSigning = disablePayloadSigning; m_archiveClasses = ParseStorageClasses(options.GetValueOrDefault(S3_ARCHIVE_CLASSES_OPTION)); } @@ -217,13 +222,20 @@ namespace Duplicati.Library.Backend public virtual async Task AddFileStreamAsync(string bucketName, string keyName, Stream source, CancellationToken cancelToken) { + (source, var md5, var tmp) = await Utility.Utility.CalculateThrottledStreamHash(source, "MD5", cancelToken).ConfigureAwait(false); + using var _ = tmp; + + md5 = Convert.ToBase64String(Utility.Utility.HexStringAsByteArray(md5)); + using var ts = source.ObserveReadTimeout(m_timeouts.ReadWriteTimeout, false); var objectAddRequest = new PutObjectRequest { BucketName = bucketName, Key = keyName, InputStream = ts, - UseChunkEncoding = m_useChunkEncoding + UseChunkEncoding = m_useChunkEncoding, + MD5Digest = md5, + DisablePayloadSigning = m_disablePayloadSigning }; if (!string.IsNullOrWhiteSpace(m_storageClass)) objectAddRequest.StorageClass = new S3StorageClass(m_storageClass); diff --git a/Duplicati/Library/Backend/S3/S3Backend.cs b/Duplicati/Library/Backend/S3/S3Backend.cs index 8d4a2be0e..b6b21c166 100644 --- a/Duplicati/Library/Backend/S3/S3Backend.cs +++ b/Duplicati/Library/Backend/S3/S3Backend.cs @@ -41,6 +41,7 @@ namespace Duplicati.Library.Backend private const string SSL_OPTION = "use-ssl"; private const string S3_CLIENT_OPTION = "s3-client"; private const string S3_DISABLE_CHUNK_ENCODING_OPTION = "s3-disable-chunk-encoding"; + private const string S3_DISABLE_PAYLOAD_SIGNING_OPTION = "s3-disable-payload-signing"; private const string S3_LIST_API_VERSION_OPTION = "s3-list-api-version"; private const string S3_RECURSIVE_LIST = "s3-recursive-list"; @@ -242,14 +243,15 @@ namespace Duplicati.Library.Backend if (!options.ContainsKey("s3-ext-forcepathstyle") && !hostname.EndsWith(".amazonaws.com", StringComparison.OrdinalIgnoreCase)) options["s3-ext-forcepathstyle"] = "true"; - var disableChunkEncoding = Utility.Utility.ParseBoolOption(options, S3_DISABLE_CHUNK_ENCODING_OPTION); var s3ClientOptionValue = options.GetValueOrDefault(S3_CLIENT_OPTION); (var awsID, var awsKey) = auth.GetCredentials(); if (string.IsNullOrWhiteSpace(s3ClientOptionValue) || string.Equals(s3ClientOptionValue, "aws", StringComparison.OrdinalIgnoreCase)) { - m_s3Client = new S3AwsClient(awsID, awsKey, locationConstraint, hostname, storageClass, useSSL, disableChunkEncoding, timeout, options); + var disableChunkEncoding = Utility.Utility.ParseBoolOption(options, S3_DISABLE_CHUNK_ENCODING_OPTION); + var disablePayloadSigning = Utility.Utility.ParseBoolOption(options, S3_DISABLE_PAYLOAD_SIGNING_OPTION); + m_s3Client = new S3AwsClient(awsID, awsKey, locationConstraint, hostname, storageClass, useSSL, disableChunkEncoding, disablePayloadSigning, timeout, options); } else if (string.Equals(s3ClientOptionValue, "minio", StringComparison.OrdinalIgnoreCase)) { @@ -336,6 +338,7 @@ namespace Duplicati.Library.Backend new CommandLineArgument(SSL_OPTION, CommandLineArgument.ArgumentType.Boolean, Strings.S3Backend.DescriptionUseSSLShort, Strings.S3Backend.DescriptionUseSSLLong), new CommandLineArgument(S3_CLIENT_OPTION, CommandLineArgument.ArgumentType.Enumeration, Strings.S3Backend.S3ClientDescriptionShort, Strings.S3Backend.S3ClientDescriptionLong, "aws", null, new string[] { "aws", "minio" }), new CommandLineArgument(S3_DISABLE_CHUNK_ENCODING_OPTION, CommandLineArgument.ArgumentType.Boolean, Strings.S3Backend.DescriptionDisableChunkEncodingShort, Strings.S3Backend.DescriptionDisableChunkEncodingLong, "false"), + new CommandLineArgument(S3_DISABLE_PAYLOAD_SIGNING_OPTION, CommandLineArgument.ArgumentType.Boolean, Strings.S3Backend.DescriptionDisablePayloadSigningShort, Strings.S3Backend.DescriptionDisablePayloadSigningLong, "false"), new CommandLineArgument(S3AwsClient.S3_ARCHIVE_CLASSES_OPTION, CommandLineArgument.ArgumentType.Flags, Strings.S3Backend.S3ArchiveClassesDescriptionShort, Strings.S3Backend.S3ArchiveClassesDescriptionLong, string.Join(",", S3AwsClient.DEFAULT_ARCHIVE_CLASSES.Select(x => x.Value)), null, KNOWN_S3_STORAGE_CLASSES.Select(x => x.Value).WhereNotNullOrWhiteSpace().ToArray()), new CommandLineArgument(S3_LIST_API_VERSION_OPTION, CommandLineArgument.ArgumentType.Enumeration, Strings.S3Backend.DescriptionListApiVersionShort, Strings.S3Backend.DescriptionListApiVersionLong, "v1", null, ["v1", "v2"]), new CommandLineArgument(S3_RECURSIVE_LIST, CommandLineArgument.ArgumentType.Boolean, Strings.S3Backend.DescriptionRecursiveListShort, Strings.S3Backend.DescriptionRecursiveListLong, "false"), diff --git a/Duplicati/Library/Utility/OverrideableStream.cs b/Duplicati/Library/Utility/OverrideableStream.cs index 35e8a5273..b331c0b2f 100644 --- a/Duplicati/Library/Utility/OverrideableStream.cs +++ b/Duplicati/Library/Utility/OverrideableStream.cs @@ -37,6 +37,11 @@ namespace Duplicati.Library.Utility /// protected Stream m_basestream; + /// + /// The base stream that is wrapped + /// + public Stream BaseStream => m_basestream; + /// /// Creates a new instance /// diff --git a/Duplicati/Library/Utility/Utility.cs b/Duplicati/Library/Utility/Utility.cs index 7f54fb5a4..d29f548ab 100644 --- a/Duplicati/Library/Utility/Utility.cs +++ b/Duplicati/Library/Utility/Utility.cs @@ -23,6 +23,7 @@ using System; using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; using System.Globalization; using System.IO; using System.Linq; @@ -1397,7 +1398,8 @@ namespace Duplicati.Library.Utility /// The wrapped commandline element. /// The argument to wrap. /// A flag indicating if environment variables are allowed to be expanded - public static string WrapCommandLineElement(string? arg, bool allowEnvExpansion) + [return: NotNullIfNotNull("arg")] + public static string? WrapCommandLineElement(string? arg, bool allowEnvExpansion) { if (string.IsNullOrWhiteSpace(arg)) return arg; @@ -1847,5 +1849,90 @@ namespace Duplicati.Library.Utility : Timeout.Infinite }; } + + /// + /// The types of streams that are considered basic (i.e. not wrapped) + /// + private static readonly IReadOnlySet _basicStreamTypes = new HashSet + { + typeof(FileStream), + typeof(MemoryStream), + typeof(NetworkStream), + typeof(BufferedStream), + typeof(System.IO.Compression.DeflateStream), + typeof(System.IO.Compression.GZipStream), + typeof(System.IO.Compression.ZLibStream) + }; + + /// + /// Unwraps a stream from layers of wrapping streams + /// + /// The stream to unwrap + /// The unwrapped stream + public static Stream UnwrapThrottledStream(this Stream stream) + { + if (stream == null) + throw new ArgumentNullException(nameof(stream)); + + var previousStream = stream; + + do + { + previousStream = stream; + + while (stream is WrappingStream wrappingStream) + stream = wrappingStream.BaseStream; + while (stream is OverrideableStream overrideableStream) + stream = overrideableStream.BaseStream; + + } while (stream != previousStream); + +#if DEBUG + if (!_basicStreamTypes.Contains(stream.GetType())) + throw new InvalidOperationException($"The unwrapped stream is not a basic stream, but a {stream.GetType()}"); +#endif + + return stream; + } + + /// + /// Calculates the hash of a throttled stream and returns the stream to read + /// + /// The source stream + /// The hash algorithm to use + /// The cancellation token to observe + /// A tuple with the stream, the hash and a temporary file if used + public static async Task<(Stream content, string hash, TempFile? tmpfile)> CalculateThrottledStreamHash(Stream stream, string hashalgorithm, CancellationToken cancelToken) + { + TempFile? tmp = null; + string contentHash; + var measure = stream.UnwrapThrottledStream(); + if (measure.CanSeek) + { + var p = measure.Position; + + // Compute the hash + using (var hashalg = HashFactory.CreateHasher(hashalgorithm)) + contentHash = ByteArrayAsHexString(hashalg.ComputeHash(measure)); + + measure.Position = p; + } + else + { + // No seeking possible, use a temp file + tmp = new TempFile(); + await using (var sr = File.OpenWrite(tmp)) + using (var hasher = HashFactory.CreateHasher(hashalgorithm)) + await using (var hc = new HashCalculatingStream(measure, hasher)) + { + await CopyStreamAsync(hc, sr, cancelToken).ConfigureAwait(false); + contentHash = hc.GetFinalHashString(); + } + + stream = File.OpenRead(tmp); + } + + return (stream, contentHash, tmp); + } } }