haeaeae/duplicati
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix throttle stream issues

Browse Source 
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.
This commit is contained in:
Kenneth Skovhede
2025-05-09 13:46:45 +02:00
parent 83c7b13f55
commit 42c3ecd009
8 changed files with 180 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Duplicati/Library/Backend/AliyunOSS/AliyunOSSBackend.cs
+6 -1
View File
@@ -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);
Duplicati/Library/Backend/Backblaze/B2.cs
+2 -41
View File
@@ -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();
}
}
/// <summary>
Duplicati/Library/Backend/Idrivee2/Idrivee2Backend.cs
+1 -2
View File
@@ -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;
Duplicati/Library/Backend/Jottacloud/Jottacloud.cs
+59 -96
View File
@@ -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 { }
}
}
}
}
Duplicati/Library/Backend/S3/S3AwsClient.cs
+14 -2
View File
@@ -61,6 +61,10 @@ namespace Duplicati.Library.Backend
/// The option to specify if chunk encoding should be used
/// </summary>
private readonly bool m_useChunkEncoding;
/// <summary>
/// The option to specify if payload signing should be disabled
/// </summary>
private readonly bool m_disablePayloadSigning;
/// <summary>
/// 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<string, string?> options)
string? storageClass, bool useSSL, bool disableChunkEncoding, bool disablePayloadSigning, TimeoutOptionsHelper.Timeouts timeouts, Dictionary<string, string?> 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);
Duplicati/Library/Backend/S3/S3Backend.cs
+5 -2
View File
@@ -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"),
Duplicati/Library/Utility/OverrideableStream.cs
+5
View File
@@ -37,6 +37,11 @@ namespace Duplicati.Library.Utility
/// </summary>
protected Stream m_basestream;
/// <summary>
/// The base stream that is wrapped
/// </summary>
public Stream BaseStream => m_basestream;
/// <summary>
/// Creates a new <see cref="OverrideableStream"/> instance
/// </summary>
Duplicati/Library/Utility/Utility.cs
+88 -1
View File
@@ -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
/// <returns>The wrapped commandline element.</returns>
/// <param name="arg">The argument to wrap.</param>
/// <param name="allowEnvExpansion">A flag indicating if environment variables are allowed to be expanded</param>
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
};
}
/// <summary>
/// The types of streams that are considered basic (i.e. not wrapped)
/// </summary>
private static readonly IReadOnlySet<Type> _basicStreamTypes = new HashSet<Type>
{
typeof(FileStream),
typeof(MemoryStream),
typeof(NetworkStream),
typeof(BufferedStream),
typeof(System.IO.Compression.DeflateStream),
typeof(System.IO.Compression.GZipStream),
typeof(System.IO.Compression.ZLibStream)
};
/// <summary>
/// Unwraps a stream from layers of wrapping streams
/// </summary>
/// <param name="stream">The stream to unwrap</param>
/// <returns>The unwrapped stream</returns>
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;
}
/// <summary>
/// Calculates the hash of a throttled stream and returns the stream to read
/// </summary>
/// <param name="stream">The source stream
/// <param name="hashalgorithm">The hash algorithm to use</param>
/// <param name="cancelToken">The cancellation token to observe</param>
/// <returns>A tuple with the stream, the hash and a temporary file if used</returns>
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);
}
}
}