#nullable enable
using Grpc.Core;
namespace Udb.Client.Generated;
public sealed record UdbCallOptions
{
public TimeSpan? Timeout { get; init; } = TimeSpan.FromSeconds(30);
public int MaxAttempts { get; init; } = 4;
public TimeSpan InitialBackoff { get; init; } = TimeSpan.FromMilliseconds(100);
public TimeSpan MaxBackoff { get; init; } = TimeSpan.FromSeconds(5);
public double BackoffMultiplier { get; init; } = 2.0;
public IReadOnlySet<StatusCode> RetryableCodes { get; init; } = new HashSet<StatusCode>
{
StatusCode.Unavailable,
StatusCode.ResourceExhausted,
};
}
public sealed class UdbRpcException : Exception
{
public StatusCode Code { get; }
public string RpcPath { get; }
public byte[]? ErrorDetail { get; }
public global::Udb.Entity.V1.ErrorDetail? DecodedErrorDetail { get; }
public bool Retryable => DecodedErrorDetail?.Retryable ?? false;
public long RetryAfterMs => DecodedErrorDetail?.RetryAfterMs ?? 0L;
public global::Udb.Entity.V1.ErrorKind Kind =>
DecodedErrorDetail?.Kind ?? global::Udb.Entity.V1.ErrorKind.Unspecified;
public System.Collections.Generic.IReadOnlyList<(string Field, string Description)> FieldViolations =>
ExtractFieldViolations(DecodedErrorDetail);
public UdbRpcException(string rpcPath, RpcException inner)
: base($"udb: RPC {rpcPath} failed with {inner.StatusCode}: {inner.Status.Detail}", inner)
{
RpcPath = rpcPath;
Code = inner.StatusCode;
ErrorDetail = ExtractDetail(inner);
DecodedErrorDetail = ParseDetail(ErrorDetail) ?? SynthesizeTransportDetail(Code);
}
private static byte[]? ExtractDetail(RpcException ex)
{
foreach (var entry in ex.Trailers)
{
if (entry.IsBinary &&
string.Equals(entry.Key, "udb-error-detail-bin", StringComparison.OrdinalIgnoreCase))
{
return entry.ValueBytes;
}
}
return null;
}
private static global::Udb.Entity.V1.ErrorDetail? ParseDetail(byte[]? raw)
{
if (raw is null || raw.Length == 0)
{
return null;
}
try
{
return global::Udb.Entity.V1.ErrorDetail.Parser.ParseFrom(raw);
}
catch
{
return null;
}
}
private static System.Collections.Generic.IReadOnlyList<(string Field, string Description)>
ExtractFieldViolations(global::Udb.Entity.V1.ErrorDetail? detail)
{
if (detail is null)
{
return System.Array.Empty<(string Field, string Description)>();
}
var field = global::Udb.Entity.V1.ErrorDetail.Descriptor.FindFieldByName("field_violations");
if (field is null || !field.IsRepeated)
{
return System.Array.Empty<(string Field, string Description)>();
}
if (field.Accessor.GetValue(detail) is not System.Collections.IEnumerable values)
{
return System.Array.Empty<(string Field, string Description)>();
}
var outValues = new System.Collections.Generic.List<(string Field, string Description)>();
foreach (var item in values)
{
if (item is not Google.Protobuf.IMessage message)
{
continue;
}
var fieldName = message.Descriptor.FindFieldByName("field");
var description = message.Descriptor.FindFieldByName("description");
outValues.Add((
fieldName?.Accessor.GetValue(message) as string ?? string.Empty,
description?.Accessor.GetValue(message) as string ?? string.Empty));
}
return outValues;
}
private static global::Udb.Entity.V1.ErrorDetail? SynthesizeTransportDetail(StatusCode code)
{
if (code is not (StatusCode.Unavailable or StatusCode.DeadlineExceeded or StatusCode.Cancelled))
{
return null;
}
return new global::Udb.Entity.V1.ErrorDetail
{
Backend = "transport",
Operation = TransportErrorOperation(code),
Retryable = code != StatusCode.Cancelled,
RetryAfterMs = 0,
Kind = global::Udb.Entity.V1.ErrorKind.Retryable,
};
}
private static string TransportErrorOperation(StatusCode code)
{
return code switch
{
StatusCode.DeadlineExceeded => "deadline_exceeded",
StatusCode.Cancelled => "cancelled",
StatusCode.Unavailable => "unavailable",
_ => "transport",
};
}
}
public abstract class GeneratedServiceBase
{
private static readonly Random Jitter = new();
protected Func<Metadata> HeadersFactory { get; }
protected UdbCallOptions Options { get; }
protected GeneratedServiceBase(Func<Metadata> headersFactory, UdbCallOptions? options)
{
HeadersFactory = headersFactory ?? throw new ArgumentNullException(nameof(headersFactory));
Options = options ?? new UdbCallOptions();
}
private CallOptions MakeCallOptions(TimeSpan? deadline, CancellationToken ct)
{
var effective = deadline ?? Options.Timeout;
DateTime? when = effective.HasValue ? DateTime.UtcNow.Add(effective.Value) : null;
return new CallOptions(HeadersFactory(), when, ct);
}
protected async Task<dynamic> InvokeUnaryAsync(
string rpcPath,
Func<CallOptions, object> call,
TimeSpan? deadline,
CancellationToken ct,
bool readOnly,
bool replaySafe,
object? request)
{
var backoff = Options.InitialBackoff;
var hasIdempotencyKey = HasIdempotencyKey(request);
for (var attempt = 1; ; attempt++)
{
try
{
var co = MakeCallOptions(deadline, ct);
dynamic rpc = call(co);
using (rpc)
{
return await rpc.ResponseAsync.ConfigureAwait(false);
}
}
catch (RpcException ex) when (
attempt < Options.MaxAttempts &&
IsRetryable(ex.StatusCode, readOnly, replaySafe, hasIdempotencyKey) &&
!ct.IsCancellationRequested)
{
await Task.Delay(NextBackoff(ref backoff), ct).ConfigureAwait(false);
}
catch (RpcException ex)
{
throw new UdbRpcException(rpcPath, ex);
}
}
}
private bool IsRetryable(StatusCode code, bool readOnly, bool replaySafe, bool hasIdempotencyKey)
{
if (readOnly)
{
return code == StatusCode.DeadlineExceeded || Options.RetryableCodes.Contains(code);
}
if (!replaySafe || !hasIdempotencyKey)
{
return false;
}
return code != StatusCode.DeadlineExceeded && Options.RetryableCodes.Contains(code);
}
private static bool HasIdempotencyKey(object? request)
{
if (request is null)
{
return false;
}
var property = request.GetType().GetProperty("IdempotencyKey");
return property?.GetValue(request) is string key && !string.IsNullOrWhiteSpace(key);
}
protected dynamic InvokeStreaming(
string rpcPath,
Func<CallOptions, object> call,
TimeSpan? deadline,
CancellationToken ct)
{
try
{
return call(MakeCallOptions(deadline, ct));
}
catch (RpcException ex)
{
throw new UdbRpcException(rpcPath, ex);
}
}
public static UdbRpcException MapStreamError(string rpcPath, RpcException ex)
=> new(rpcPath, ex);
private TimeSpan NextBackoff(ref TimeSpan backoff)
{
var current = backoff;
var nextMs = Math.Min(
backoff.TotalMilliseconds * Options.BackoffMultiplier,
Options.MaxBackoff.TotalMilliseconds);
backoff = TimeSpan.FromMilliseconds(nextMs);
var jittered = current.TotalMilliseconds * (0.5 + Jitter.NextDouble() * 0.5);
return TimeSpan.FromMilliseconds(jittered);
}
}