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

Refactor the Retry-After logic. Now it tracks Retry-After headers on all requests, and blocks any future requests until the time from the prior request has passed.

Browse Source
This commit is contained in:
Tyler Gill
2021-03-16 11:40:38 -06:00
parent 7894152489
commit b0793d3e49
2 changed files with 85 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Duplicati/Library/Backend/OneDrive/Exceptions.cs
-23
View File
@@ -81,29 +81,6 @@ namespace Duplicati.Library.Backend.MicrosoftGraph
}
}
public RetryConditionHeaderValue RetryAfter
{
get
{
if (this.responseMessage != null)
{
return this.responseMessage.Headers.RetryAfter;
}
else
{
string header = this.webResponse.Headers[HttpResponseHeader.RetryAfter];
if (header != null && RetryConditionHeaderValue.TryParse(header, out RetryConditionHeaderValue retryAfter))
{
return retryAfter;
}
else
{
return null;
}
}
}
}
protected static string ResponseToString(HttpResponseMessage response)
{
if (response != null)
Duplicati/Library/Backend/OneDrive/MicrosoftGraphBackend.cs
+85 -63
View File
@@ -99,6 +99,11 @@ namespace Duplicati.Library.Backend
private readonly Lazy<string> rootPathFromURL;
private string RootPath => this.rootPathFromURL.Value;
// Whenever a response includes a Retry-After header, we'll update this timestamp with when we can next
// send a request. And before sending any requests, we'll make sure to wait until at least this time.
// Since this may be read and written by multiple threads, it is stored as a long and updated using Interlocked.Exchange.
private long retryAfter = DateTimeOffset.MinValue.UtcTicks;
protected MicrosoftGraphBackend() { } // Constructor needed for dynamic loading to find it
protected MicrosoftGraphBackend(string url, string protocolKey, Dictionary<string, string> options)
@@ -209,6 +214,7 @@ namespace Duplicati.Library.Backend
UploadSession uploadSession = this.Post<UploadSession>(string.Format("{0}/root:{1}{2}:/createUploadSession", this.DrivePrefix, this.RootPath, NormalizeSlashes(dnsTestFile)), MicrosoftGraphBackend.dummyUploadSession);
// Canceling an upload session is done by sending a DELETE to the upload URL
this.WaitForRetryAfter();
if (this.m_client != null)
{
using (var request = new HttpRequestMessage(HttpMethod.Delete, uploadSession.UploadUrl))
@@ -354,6 +360,7 @@ namespace Duplicati.Library.Backend
{
try
{
this.WaitForRetryAfter();
string getUrl = string.Format("{0}/root:{1}{2}:/content", this.DrivePrefix, this.RootPath, NormalizeSlashes(remotename));
if (this.m_client != null)
{
@@ -411,6 +418,7 @@ namespace Duplicati.Library.Backend
// PUT only supports up to 4 MB file uploads. There's a separate process for larger files.
if (stream.Length < PUT_MAX_SIZE)
{
await this.WaitForRetryAfter(cancelToken).ConfigureAwait(false);
string putUrl = string.Format("{0}/root:{1}{2}:/content", this.DrivePrefix, this.RootPath, NormalizeSlashes(remotename));
if (this.m_client != null)
{
@@ -443,6 +451,7 @@ namespace Duplicati.Library.Backend
string createSessionUrl = string.Format("{0}/root:{1}{2}:/createUploadSession", this.DrivePrefix, this.RootPath, NormalizeSlashes(remotename));
if (this.m_client != null)
{
await this.WaitForRetryAfter(cancelToken).ConfigureAwait(false);
using (HttpRequestMessage createSessionRequest = new HttpRequestMessage(HttpMethod.Post, createSessionUrl))
using (HttpResponseMessage createSessionResponse = await this.m_client.SendAsync(createSessionRequest, cancelToken).ConfigureAwait(false))
{
@@ -461,6 +470,7 @@ namespace Duplicati.Library.Backend
int retryCount = this.fragmentRetryCount;
for (int attempt = 0; attempt < retryCount; attempt++)
{
await this.WaitForRetryAfter(cancelToken).ConfigureAwait(false);
using (HttpRequestMessage request = new HttpRequestMessage(HttpMethod.Put, uploadSession.UploadUrl))
using (StreamContent fragmentContent = new StreamContent(subStream))
{
@@ -498,7 +508,7 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
}
@@ -513,23 +523,14 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
else if ((int)ex.StatusCode >= 500 && (int)ex.StatusCode < 600)
{
// If there was a Retry-After header, then wait for that amount of time first.
// Interestingly, this isn't mentioned in the recommendations above, but is documented
// in some of the other OneDrive general docs.
bool waited = await this.WaitForRetryAfter(ex.RetryAfter, cancelToken);
// If we didn't wait due to a Retry-After header, wait using the strategy defined in the recommendations.
if (!waited)
{
// If a 5xx error code is hit, we should use an exponential backoff strategy before retrying.
// To make things simpler, we just use the current attempt number as the exponential factor.
await Task.Delay((int)Math.Pow(2, attempt) * this.fragmentRetryDelay);
}
// If a 5xx error code is hit, we should use an exponential backoff strategy before retrying.
// To make things simpler, we just use the current attempt number as the exponential factor.
// If there was a Retry-After header, we'll wait for that right before sending the next request as well.
await Task.Delay((int)Math.Pow(2, attempt) * this.fragmentRetryDelay).ConfigureAwait(false);
continue;
}
else if (ex.StatusCode == HttpStatusCode.NotFound)
@@ -542,13 +543,11 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
else if ((int)ex.StatusCode >= 400 && (int)ex.StatusCode < 500)
{
// If a 4xx error code is hit, we should retry without the exponential backoff attempt.
// However, we should still wait for any Retry-After header limits.
await this.WaitForRetryAfter(ex.RetryAfter, cancelToken);
continue;
}
else
@@ -560,7 +559,7 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
}
catch (Exception ex)
@@ -572,7 +571,7 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
// If we successfully sent this piece, then we can break out of the retry loop
@@ -585,7 +584,8 @@ namespace Duplicati.Library.Backend
}
else
{
using (HttpWebResponse createSessionResponse = await this.m_oAuthHelper.GetResponseWithoutExceptionAsync(createSessionUrl, cancelToken, MicrosoftGraphBackend.dummyUploadSession, HttpMethod.Post.ToString()))
await this.WaitForRetryAfter(cancelToken).ConfigureAwait(false);
using (HttpWebResponse createSessionResponse = await this.m_oAuthHelper.GetResponseWithoutExceptionAsync(createSessionUrl, cancelToken, MicrosoftGraphBackend.dummyUploadSession, HttpMethod.Post.ToString()).ConfigureAwait(false))
{
UploadSession uploadSession = this.ParseResponse<UploadSession>(createSessionResponse);
@@ -602,6 +602,8 @@ namespace Duplicati.Library.Backend
int retryCount = this.fragmentRetryCount;
for (int attempt = 0; attempt < retryCount; attempt++)
{
await this.WaitForRetryAfter(cancelToken).ConfigureAwait(false);
// The uploaded put requests will error if they are authenticated
var request = new AsyncHttpRequest(this.m_oAuthHelper.CreateRequest(uploadSession.UploadUrl, HttpMethod.Put.ToString(), true));
request.Request.ContentLength = read;
@@ -610,12 +612,12 @@ namespace Duplicati.Library.Backend
using (var requestStream = request.GetRequestStream(read))
{
await Utility.Utility.CopyStreamAsync(subStream, requestStream, cancelToken);
await Utility.Utility.CopyStreamAsync(subStream, requestStream, cancelToken).ConfigureAwait(false);
}
try
{
using (var response = await this.m_oAuthHelper.GetResponseWithoutExceptionAsync(request, cancelToken))
using (var response = await this.m_oAuthHelper.GetResponseWithoutExceptionAsync(request, cancelToken).ConfigureAwait(false))
{
// Note: On the last request, the json result includes the default properties of the item that was uploaded
this.ParseResponse<UploadSession>(response);
@@ -641,7 +643,7 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
}
@@ -656,23 +658,14 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
else if ((int)ex.StatusCode >= 500 && (int)ex.StatusCode < 600)
{
// If there was a Retry-After header, then wait for that amount of time first.
// Interestingly, this isn't mentioned in the recommendations above, but is documented
// in some of the other OneDrive general docs.
bool waited = await this.WaitForRetryAfter(ex.RetryAfter, cancelToken);
// If we didn't wait due to a Retry-After header, wait using the strategy defined in the recommendations.
if (!waited)
{
// If a 5xx error code is hit, we should use an exponential backoff strategy before retrying.
// To make things simpler, we just use the current attempt number as the exponential factor.
await Task.Delay((int)Math.Pow(2, attempt) * this.fragmentRetryDelay);
}
// If a 5xx error code is hit, we should use an exponential backoff strategy before retrying.
// To make things simpler, we just use the current attempt number as the exponential factor.
// If there was a Retry-After header, we'll wait for that right before sending the next request as well.
await Task.Delay((int)Math.Pow(2, attempt) * this.fragmentRetryDelay).ConfigureAwait(false);
continue;
}
else if (ex.StatusCode == HttpStatusCode.NotFound)
@@ -685,13 +678,11 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
else if ((int)ex.StatusCode >= 400 && (int)ex.StatusCode < 500)
{
// If a 4xx error code is hit, we should retry without the exponential backoff attempt.
// However, we should still wait for any Retry-After header limits.
await this.WaitForRetryAfter(ex.RetryAfter, cancelToken);
continue;
}
else
@@ -703,7 +694,7 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
}
catch (Exception ex)
@@ -715,7 +706,7 @@ namespace Duplicati.Library.Backend
(int)(offset / bufferSize),
(int)Math.Ceiling((double)stream.Length / bufferSize),
ex,
cancelToken);
cancelToken).ConfigureAwait(false);
}
// If we successfully sent this piece, then we can break out of the retry loop
@@ -732,6 +723,7 @@ namespace Duplicati.Library.Backend
{
try
{
this.WaitForRetryAfter();
string deleteUrl = string.Format("{0}/root:{1}{2}", this.DrivePrefix, this.RootPath, NormalizeSlashes(remotename));
if (this.m_client != null)
{
@@ -811,6 +803,7 @@ namespace Duplicati.Library.Backend
}
else
{
this.WaitForRetryAfter();
using (var response = this.m_oAuthHelper.GetResponseWithoutException(url, null, method.ToString()))
{
return this.ParseResponse<T>(response);
@@ -830,6 +823,7 @@ namespace Duplicati.Library.Backend
}
else
{
this.WaitForRetryAfter();
using (var response = this.m_oAuthHelper.GetResponseWithoutException(url, body, method.ToString()))
{
return this.ParseResponse<T>(response);
@@ -839,6 +833,7 @@ namespace Duplicati.Library.Backend
private T SendRequest<T>(HttpRequestMessage request)
{
this.WaitForRetryAfter();
using (var response = this.m_client.SendAsync(request).Await())
{
return this.ParseResponse<T>(response);
@@ -872,6 +867,8 @@ namespace Duplicati.Library.Backend
private void CheckResponse(HttpResponseMessage response)
{
this.SetRetryAfter(response.Headers.RetryAfter);
if (!response.IsSuccessStatusCode)
{
if (response.StatusCode == HttpStatusCode.NotFound)
@@ -889,6 +886,12 @@ namespace Duplicati.Library.Backend
private void CheckResponse(HttpWebResponse response)
{
string retryAfterHeader = response.Headers[HttpResponseHeader.RetryAfter];
if (retryAfterHeader != null && RetryConditionHeaderValue.TryParse(retryAfterHeader, out RetryConditionHeaderValue retryAfter))
{
this.SetRetryAfter(retryAfter);
}
if (!((int)response.StatusCode >= 200 && (int)response.StatusCode < 300))
{
if (response.StatusCode == HttpStatusCode.NotFound)
@@ -957,7 +960,7 @@ namespace Duplicati.Library.Backend
// Before throwing the exception, cancel the upload session
// The uploaded delete request will error if it is authenticated
var request = new AsyncHttpRequest(this.m_oAuthHelper.CreateRequest(uploadSession.UploadUrl, HttpMethod.Delete.ToString(), true));
using (var response = await this.m_oAuthHelper.GetResponseWithoutExceptionAsync(request, cancelToken))
using (var response = await this.m_oAuthHelper.GetResponseWithoutExceptionAsync(request, cancelToken).ConfigureAwait(false))
{
// Note that the response body should always be empty in this case.
this.ParseResponse<UploadSession>(response);
@@ -966,39 +969,58 @@ namespace Duplicati.Library.Backend
throw new UploadSessionException(createSessionResponse, fragment, fragmentCount, ex);
}
private async Task<bool> WaitForRetryAfter(RetryConditionHeaderValue retryAfter, CancellationToken cancelToken)
private void SetRetryAfter(RetryConditionHeaderValue retryAfter)
{
if (retryAfter != null)
{
TimeSpan delay = TimeSpan.Zero;
DateTimeOffset? delayUntil = null;
if (retryAfter.Delta.HasValue)
{
delay = retryAfter.Delta.Value;
delayUntil = DateTimeOffset.UtcNow + retryAfter.Delta.Value;
}
else if (retryAfter.Date.HasValue)
{
DateTimeOffset now = DateTimeOffset.UtcNow;
DateTimeOffset delayUntil = retryAfter.Date.Value;
// Make sure delayUntil is in the future
if (delayUntil >= now)
{
delay = now - delayUntil;
}
else
{
// If the date given was in the past then don't wait at all
}
delayUntil = retryAfter.Date.Value;
}
if (delay != TimeSpan.Zero)
if (delayUntil.HasValue)
{
await Task.Delay(delay);
return true;
// Set the retry timestamp to the UTC version of the timestamp.
Interlocked.Exchange(ref this.retryAfter, delayUntil.Value.UtcTicks);
}
}
}
return false;
private void WaitForRetryAfter()
{
this.WaitForRetryAfter(CancellationToken.None).Await();
}
private async Task WaitForRetryAfter(CancellationToken cancelToken)
{
// Make sure this is thread safe in case multiple calls are made concurrently to this backend
// This is done by reading the value into a local value which is then parsed and operated on locally.
long retryAfterTicks = Interlocked.Read(ref this.retryAfter);
DateTimeOffset delayUntil = new DateTimeOffset(retryAfterTicks, TimeSpan.Zero);
TimeSpan delay;
DateTimeOffset now = DateTimeOffset.UtcNow;
// Make sure delayUntil is in the future and delay until then if so
if (delayUntil >= now)
{
delay = delayUntil - now;
}
else
{
// If the date given was in the past then don't wait at all
delay = TimeSpan.Zero;
}
if (delay > TimeSpan.Zero)
{
await Task.Delay(delay).ConfigureAwait(false);
}
}
/// <summary>