Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/CoreSync.Http.Client/CoreSync.Http.Client.csproj
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@

<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Http" Version="10.0.9" />
<PackageReference Include="MessagePack" Version="3.1.7" />
<PackageReference Include="MessagePack" Version="3.1.12" />
<PackageReference Include="Polly.Core" Version="8.7.0" />
</ItemGroup>

Expand Down
130 changes: 110 additions & 20 deletions src/CoreSync.Http.Client/Implementation/SyncProviderHttpClient.cs
Original file line number Diff line number Diff line change
Expand Up @@ -34,18 +34,34 @@ public SyncProviderHttpClient(IHttpClientFactory httpClientProvider, SyncProvide

/// <summary>
/// Throws for an unsuccessful response, rebuilding <see cref="SyncAnchorTooOldException"/> from the
/// server's 410 Gone signal so that callers see the same typed error they would from a local provider.
/// server's 410 Gone signal and <see cref="SyncHttpRequestException"/> from everything else, so that
/// callers see what the server said instead of a bare status code.
/// </summary>
/// <remarks>
/// <para>
/// Used in place of <see cref="HttpResponseMessage.EnsureSuccessStatusCode"/> on every sync endpoint:
/// without it an aged-out anchor arrives as a bare <see cref="HttpRequestException"/> and the only
/// thing an application can tell the user is that synchronization failed.
/// without it a failure arrives as a bare <see cref="HttpRequestException"/> and the only thing an
/// application can tell the user is that synchronization failed.
/// </para>
/// <para>
/// <see cref="SyncHttpRequestException"/> derives from <see cref="HttpRequestException"/> and carries
/// the very message <see cref="HttpResponseMessage.EnsureSuccessStatusCode"/> would have produced, so
/// the failure callers used to see is still there; the <see cref="SyncHttpHeaders.ErrorCode"/> header
/// and <see cref="SyncErrorDetails"/> body are read when present and inferred from the status code
/// when the server (an older one) sends neither.
/// </para>
/// </remarks>
private static async Task EnsureSyncSuccessAsync(HttpResponseMessage response, CancellationToken cancellationToken)
{
if (response.StatusCode == HttpStatusCode.Gone &&
response.Headers.TryGetValues(SyncHttpHeaders.ErrorCode, out var errorCodes) &&
errorCodes.Contains(SyncHttpErrorCodes.AnchorTooOld))
if (response.IsSuccessStatusCode)
{
return;
}

var errorCode = ReadHeader(response, SyncHttpHeaders.ErrorCode);
var traceId = ReadHeader(response, SyncHttpHeaders.TraceId);

if (response.StatusCode == HttpStatusCode.Gone && errorCode == SyncHttpErrorCodes.AnchorTooOld)
{
SyncAnchorTooOldError? error = null;
try
Expand All @@ -66,9 +82,62 @@ private static async Task EnsureSyncSuccessAsync(HttpResponseMessage response, C
: new SyncAnchorTooOldException(requestedVersion, minValidVersion);
}

response.EnsureSuccessStatusCode();
// The body is only read on the server's word: a JSON body without the header is the host's
// (a ProblemDetails for a 404, an error page), and reading it as SyncErrorDetails would
// invent an error code the server never sent.
SyncErrorDetails? details = null;
if (errorCode != null && IsJson(response.Content.Headers.ContentType))
{
try
{
details = await response.Content.ReadFromJsonAsync<SyncErrorDetails>(cancellationToken);
}
catch (Exception ex) when (ex is JsonException || ex is NotSupportedException)
{
// A body in a shape we do not know is not a reason to lose the header or the status code.
}
}

string message;
try
{
response.EnsureSuccessStatusCode();
return;
}
catch (HttpRequestException ex)
{
// Reused verbatim so the message callers have been reading (and logging) does not change.
message = ex.Message;
}

throw new SyncHttpRequestException(message, response.StatusCode, errorCode, traceId, details);
}

private static string? ReadHeader(HttpResponseMessage response, string name)
=> response.Headers.TryGetValues(name, out var values) ? values.FirstOrDefault() : null;

private static bool IsJson(MediaTypeHeaderValue? contentType)
=> contentType?.MediaType is { } mediaType
&& (mediaType.Equals("application/json", StringComparison.OrdinalIgnoreCase)
|| mediaType.EndsWith("+json", StringComparison.OrdinalIgnoreCase));

/// <summary>
/// Decides whether a failed bulk download request is worth repeating.
/// </summary>
/// <remarks>
/// Only a failure the server marked as permanent is not retried: an aged-out anchor, an expired
/// session, a data or configuration error. Everything else - a dropped connection, a timeout, a
/// 5xx without an error code from an older server or a proxy - keeps the retries it always had,
/// because those are what let a synchronization survive a flaky connection in the field.
/// </remarks>
private static bool IsWorthRetrying(Exception exception)
=> exception switch
{
SyncAnchorTooOldException => false,
SyncHttpRequestException http => http.IsTransient,
_ => true
};

public async Task<SyncAnchor> ApplyChangesAsync([NotNull] SyncChangeSet changeSet, [CanBeNull] Func<SyncItem, ConflictResolution>? onConflictFunc = null, CancellationToken cancellationToken = default)
{
var httpClient = _httpClientFactory.CreateClient(_options.HttpClientName ?? Options.DefaultName);
Expand Down Expand Up @@ -132,7 +201,7 @@ await httpClient.PostAsync($"{_options.SyncControllerRoute}/changes-bulk-item",
SyncProgress?.Invoke(this, new SyncProgressEventArgs(SyncStage.ApplyChanges, skip / (double)changeSet.Items.Count));
}

var remoteChangeSetResponse = await httpClient.PostAsync(
using var remoteChangeSetResponse = await httpClient.PostAsync(
$"{_options.SyncControllerRoute}/changes-bulk-complete{(_options.UseBinaryFormat ? "-binary" : string.Empty)}/{sessionId}", null, cancellationToken);
await EnsureSyncSuccessAsync(remoteChangeSetResponse, cancellationToken);

Expand All @@ -154,7 +223,7 @@ public async Task<SyncChangeSet> GetChangesAsync(Guid otherStoreId, SyncFilterPa
request.Headers.Add(SyncHttpHeaders.TablesCount, tables.Length.ToString());
}

var response = await httpClient.SendAsync(request, cancellationToken);
using var response = await httpClient.SendAsync(request, cancellationToken);
await EnsureSyncSuccessAsync(response, cancellationToken);

var bulkSyncChangeSet = await response.Content.ReadFromJsonAsync<BulkSyncChangeSet>(cancellationToken)
Expand All @@ -177,15 +246,30 @@ private async Task<SyncChangeSet> DownloadBulkChangeSetJson(BulkSyncChangeSet bu
SyncProgress?.Invoke(this, new SyncProgressEventArgs(stage, 0.0));

ResiliencePipeline pipeline = new ResiliencePipelineBuilder()
.AddRetry(new RetryStrategyOptions { MaxRetryAttempts = _options.MaxRetryAttempts }) // Add retry using the default options
.AddRetry(new RetryStrategyOptions
{
MaxRetryAttempts = _options.MaxRetryAttempts,
// Polly's default handles every exception except a cancellation; keep that and
// additionally stop on failures the server marked as permanent.
ShouldHandle = args => new ValueTask<bool>(
args.Outcome.Exception is { } exception
&& exception is not OperationCanceledException
&& IsWorthRetrying(exception))
})
.Build(); // Builds the resilience pipeline

List<SyncItem> items = [];
for (var skip = 0; skip < bulkSyncChangeSet.TotalChanges; skip += _options.BulkItemSize)
{
await pipeline.ExecuteAsync(async (token) =>
{
var bulkItems = await httpClient.GetFromJsonAsync<SyncItem[]>($"/{_options.SyncControllerRoute}/changes-bulk-item/{bulkSyncChangeSet.SessionId}/{skip}/{_options.BulkItemSize}", cancellationToken: token)
// Headers first and disposed when done, as GetFromJsonAsync did before the response had
// to be inspected for the error signal.
using var response = await httpClient.GetAsync($"/{_options.SyncControllerRoute}/changes-bulk-item/{bulkSyncChangeSet.SessionId}/{skip}/{_options.BulkItemSize}", HttpCompletionOption.ResponseHeadersRead, token);

await EnsureSyncSuccessAsync(response, token);

var bulkItems = await response.Content.ReadFromJsonAsync<SyncItem[]>(cancellationToken: token)
?? throw new InvalidOperationException();

items.AddRange(bulkItems);
Expand Down Expand Up @@ -219,7 +303,7 @@ private async Task<SyncChangeSet> DownloadBulkChangeSetBinary(BulkSyncChangeSet

try
{
var response = await httpClient.GetAsync($"{_options.SyncControllerRoute}/changes-bulk-item-binary/{bulkSyncChangeSet.SessionId}/{skip}/{_options.BulkItemSize}", cancellationToken);
using var response = await httpClient.GetAsync($"{_options.SyncControllerRoute}/changes-bulk-item-binary/{bulkSyncChangeSet.SessionId}/{skip}/{_options.BulkItemSize}", HttpCompletionOption.ResponseHeadersRead, cancellationToken);

await EnsureSyncSuccessAsync(response, cancellationToken);

Expand All @@ -229,10 +313,11 @@ private async Task<SyncChangeSet> DownloadBulkChangeSetBinary(BulkSyncChangeSet
items.AddRange(bulkItems);
break;
}
catch (SyncAnchorTooOldException)
catch (Exception ex) when (!IsWorthRetrying(ex))
{
// Permanent for this client: the change history it asked for is gone. Retrying
// burns three round trips and fifteen seconds to reach the same answer.
// Permanent for this client (the change history it asked for is gone, the session
// expired, the server refused the data). Retrying burns three round trips and
// fifteen seconds to reach the same answer.
throw;
}
catch (Exception)
Expand Down Expand Up @@ -301,26 +386,31 @@ private static void ConvertJsonValueToNetObject(SyncChangeSet changeSet)
public async Task<Guid> GetStoreIdAsync(CancellationToken cancellationToken = default)
{
var httpClient = _httpClientFactory.CreateClient(_options.HttpClientName ?? Options.DefaultName);
var res = await httpClient.GetAsync($"/{_options.SyncControllerRoute}/store-id", cancellationToken);
using var res = await httpClient.GetAsync($"/{_options.SyncControllerRoute}/store-id", cancellationToken);

// Kept ahead of the generic handling on purpose: applications have been catching this exact
// type around SynchronizeAsync to trigger a re-authentication.
if (res.StatusCode == HttpStatusCode.Unauthorized)
throw new UnauthorizedAccessException();

res.EnsureSuccessStatusCode();
await EnsureSyncSuccessAsync(res, cancellationToken);

return Guid.Parse(await res.Content.ReadAsStringAsync(cancellationToken));
}

public async Task SaveVersionForStoreAsync(Guid otherStoreId, long version, CancellationToken cancellationToken = default)
{
var httpClient = _httpClientFactory.CreateClient(_options.HttpClientName ?? Options.DefaultName);
(await httpClient.PostAsync($"/{_options.SyncControllerRoute}/save-version/{otherStoreId}/{version}", null, cancellationToken))
.EnsureSuccessStatusCode();
using var response = await httpClient.PostAsync($"/{_options.SyncControllerRoute}/save-version/{otherStoreId}/{version}", null, cancellationToken);
await EnsureSyncSuccessAsync(response, cancellationToken);
}

public async Task<SyncVersion> GetSyncVersionAsync(CancellationToken cancellationToken = default)
{
var httpClient = _httpClientFactory.CreateClient(_options.HttpClientName ?? Options.DefaultName);
return await httpClient.GetFromJsonAsync<SyncVersion>($"/{_options.SyncControllerRoute}/sync-version", cancellationToken) ?? throw new InvalidOperationException();
using var response = await httpClient.GetAsync($"/{_options.SyncControllerRoute}/sync-version", HttpCompletionOption.ResponseHeadersRead, cancellationToken);
await EnsureSyncSuccessAsync(response, cancellationToken);

return await response.Content.ReadFromJsonAsync<SyncVersion>(cancellationToken) ?? throw new InvalidOperationException();
}
}
120 changes: 120 additions & 0 deletions src/CoreSync.Http.Client/SyncHttpRequestException.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
using System;
using System.Net;
using System.Net.Http;

namespace CoreSync.Http.Client;

/// <summary>
/// Raised by <see cref="ISyncProviderHttpClient"/> when the server answers a synchronization request
/// with a non-success status code, carrying whatever the server said about the failure.
/// </summary>
/// <remarks>
/// <para>
/// It derives from <see cref="HttpRequestException"/> and keeps the message and
/// <see cref="HttpRequestException.StatusCode"/> that <see cref="HttpResponseMessage.EnsureSuccessStatusCode"/>
/// produces, so code written against the previous behaviour - typically
/// <c>catch (SynchronizationException ex) when (ex.InnerException is HttpRequestException)</c> - keeps
/// working unchanged. What is new is the classification: <see cref="Kind"/>, <see cref="ErrorCode"/>,
/// <see cref="TableName"/>, <see cref="TraceId"/> and <see cref="ServerMessage"/>, read from the
/// <see cref="SyncHttpHeaders.ErrorCode"/> header and <see cref="SyncErrorDetails"/> body when the
/// server sends them, inferred from the status code when it does not (an older server).
/// </para>
/// <para>
/// A genuine connectivity failure (no route, DNS, TLS, connection reset) is not this exception: it
/// stays the plain <see cref="HttpRequestException"/> that <see cref="HttpClient"/> throws, with no
/// status code, which is how <see cref="SyncAgent"/> tells "the server was never reached" from "the
/// server answered with an error".
/// </para>
/// </remarks>
public class SyncHttpRequestException : HttpRequestException, ISyncFailureInfo
{
/// <summary>
/// Initializes a new instance of the <see cref="SyncHttpRequestException"/> class.
/// </summary>
/// <param name="message">
/// The message; callers pass the one <see cref="HttpResponseMessage.EnsureSuccessStatusCode"/>
/// would have produced, so that it stays identical to the previous behaviour.
/// </param>
/// <param name="statusCode">The response status code.</param>
/// <param name="errorCode">The <see cref="SyncHttpHeaders.ErrorCode"/> header value, or <c>null</c>.</param>
/// <param name="traceId">The <see cref="SyncHttpHeaders.TraceId"/> header value, or <c>null</c>.</param>
/// <param name="details">The parsed <see cref="SyncErrorDetails"/> body, or <c>null</c>.</param>
public SyncHttpRequestException(
string message,
HttpStatusCode statusCode,
string? errorCode,
string? traceId,
SyncErrorDetails? details)
: base(message, null, statusCode)
{
ErrorCode = errorCode;
TraceId = traceId ?? details?.TraceId;
TableName = details?.TableName;
ServerMessage = details?.Message;
Detail = details?.Detail;

// The header is the server's word and wins. Without it the status code is all there is, and
// the mapping is deliberately coarse: the point is to tell "the server answered with an error"
// (so the device is online) from everything else, not to guess at causes.
Kind = ErrorCode != null
? SyncHttpErrorCodes.ToFailureKind(ErrorCode)
: statusCode switch
{
HttpStatusCode.Unauthorized => SyncFailureKind.Unauthorized,
HttpStatusCode.Forbidden => SyncFailureKind.Unauthorized,
HttpStatusCode.RequestTimeout => SyncFailureKind.ServerUnavailable,
HttpStatusCode.TooManyRequests => SyncFailureKind.ServerUnavailable,
HttpStatusCode.BadGateway => SyncFailureKind.ServerUnavailable,
HttpStatusCode.ServiceUnavailable => SyncFailureKind.ServerUnavailable,
HttpStatusCode.GatewayTimeout => SyncFailureKind.ServerUnavailable,
_ => SyncFailureKind.ServerError
};

// Only a server that says "permanent" stops a retry. A server that says nothing (older
// versions, a proxy's own error page) gets the retries it always got, because those are what
// rescue a sync over a flaky connection.
IsTransient = !SyncHttpErrorCodes.IsPermanent(ErrorCode);

SetData(SynchronizationException.DataKeyKind, Kind.ToString());
SetData(SynchronizationException.DataKeyErrorCode, ErrorCode);
SetData(SynchronizationException.DataKeyTableName, TableName);
SetData(SynchronizationException.DataKeyTraceId, TraceId);
SetData(SynchronizationException.DataKeyServerMessage, ServerMessage);
SetData("CoreSync.StatusCode", ((int)statusCode).ToString());
}

/// <inheritdoc />
public SyncFailureKind Kind { get; }

/// <inheritdoc />
public string? ErrorCode { get; }

/// <inheritdoc />
public string? TableName { get; }

/// <inheritdoc />
public string? TraceId { get; }

/// <inheritdoc />
public string? ServerMessage { get; }

/// <summary>
/// Gets the diagnostic detail the server attached (exception types and messages), when the server
/// is configured to include it; <c>null</c> otherwise.
/// </summary>
public string? Detail { get; }

/// <summary>
/// Gets a value indicating whether repeating the same request could succeed. <c>false</c> only
/// when the server marked the failure as permanent through its error code.
/// </summary>
public bool IsTransient { get; }

private void SetData(string key, string? value)
{
if (value != null)
{
Data[key] = value;
}
}
}
2 changes: 1 addition & 1 deletion src/CoreSync.Http.Server/CoreSync.Http.Server.csproj
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@

<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App" />
<PackageReference Include="MessagePack.AspNetCoreMvcFormatter" Version="3.1.7" />
<PackageReference Include="MessagePack.AspNetCoreMvcFormatter" Version="3.1.12" />
</ItemGroup>

<ItemGroup>
Expand Down
Loading
Loading