2601 - Cannot insert duplicate key row in object.
Our Policies
We'll want to handle these errors differently and execute different paths depending on the type of error received. One policy won't do. Let's take a look at what we want our policies to behave like:
First In Line, Timeout
We want to set up a timeout for all our actions to take place; Dapper has the ability to set timeouts per command but we want a global timeout on our entire request. Polly makes this easy. For this, we setup a simple Timeout Async policy
var timeoutPolicy = Policy.TimeoutAsync(maxTimeout ?? TimeSpan.FromMinutes(2));
Next, We Handle Transient Errors
For our transient errors we want to retry and to log a warning with the associated sql and parameters that were associated with the transient error. We do this by utilizing the execution Context that we'll describe more later and set up a Wait And Retry Async policy.
var transientPolicy = Policy.Handle<SqlException>(ex => transientNumbers.Contains(ex.Number))
.WaitAndRetryAsync(
transientRetries,
attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
(ex, _, ctx) => ctx.GetLogger()?.LogWarning(ex, "{@Operation} Encountered Transient SqlException. Params:{@Param} Sql:{@Sql}", ctx.OperationKey, ctx[ParamContextKey], ctx[SqlContextKey]));
Next, The Networking Errors
For networking errors we want another Wait And Retry Async policy and not only log, as we did with transient error, but we also want to clear the connection pool in case it's become corrupted. We do this again by accessing available data within our Context.
var networkPolicy = Policy.Handle<SqlException>(ex => networkingNumbers.Contains(ex.Number))
.WaitAndRetryAsync(
networkRetries,
attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
(ex, _, ctx) =>
{
ctx.GetLogger()?.LogWarning(ex, "{@Operation} Encountered a Network Error. Params:{@Param} Sql:{@Sql}", ctx.OperationKey, ctx[ParamContextKey], ctx[SqlContextKey]);
if (ctx.TryGetConnection(out var connection))
SqlConnection.ClearPool(connection);
});
Next, Constraint Violations
If we violate a constraint with a combination of query and parameters a retry is not going to solve the problem. In this case we create a Circuit Breaker that opens and never closes again, lggoing and returning the error straight back to the consumer.
var constraintPolicy = Policy.Handle<SqlException>(ex => constraintViolationNumbers.Contains(ex.Number))
.CircuitBreakerAsync(
1,
TimeSpan.MaxValue,
(ex, _, ctx) => ctx.GetLogger()?.LogError(ex, "{@Operation} Encountered a Constraint Violation. Params:{@Param} Sql:{@Sql}", ctx.OperationKey, ctx[ParamContextKey], ctx[SqlContextKey]),
ctx => { }
);
Putting The Policies Together
Finally we wrap our policies in our desired execution order and return our fully formed resiliency policy.
var resiliencyPolicy = timeoutPolicy
.WrapAsync(transientPolicy)
.WrapAsync(networkPolicy)
.WrapAsync(constraintPolicy);
return resiliencyPolicy;
All Together Now
And here's what all this together looks like.
public static class SqlResiliencyPolicy
{
private static readonly ISet<int> transientNumbers = new HashSet<int>(new[] { 40613, 40197, 40501, 49918, 40549, 40550, 1205 });
private static readonly ISet<int> networkingNumbers = new HashSet<int>(new[] { 258, -2, 10060, 0, 64, 26, 40, 10053 });
private static readonly ISet<int> constraintViolationNumbers = new HashSet<int>(new[] { 2627, 547, 2601 });
public static IAsyncPolicy GetSqlResiliencyPolicy(TimeSpan? maxTimeout = null, int transientRetries = 3, int networkRetries = 3)
{
var timeoutPolicy = Policy.TimeoutAsync(maxTimeout ?? TimeSpan.FromMinutes(2));
var transientPolicy = Policy.Handle<SqlException>(ex => transientNumbers.Contains(ex.Number))
.WaitAndRetryAsync(
transientRetries,
attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
(ex, _, ctx) => ctx.GetLogger()?.LogWarning(ex, "{@Operation} Encountered Transient SqlException. Params:{@Param} Sql:{@Sql}", ctx.OperationKey, ctx[ParamContextKey], ctx[SqlContextKey]));
var networkPolicy = Policy.Handle<SqlException>(ex => networkingNumbers.Contains(ex.Number))
.WaitAndRetryAsync(
networkRetries,
attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
(ex, _, ctx) =>
{
ctx.GetLogger()?.LogWarning(ex, "{@Operation} Encountered a Network Error. Params:{@Param} Sql:{@Sql}", ctx.OperationKey, ctx[ParamContextKey], ctx[SqlContextKey]);
if (ctx.TryGetConnection(out var connection))
SqlConnection.ClearPool(connection);
});
var constraintPolicy = Policy.Handle<SqlException>(ex => constraintViolationNumbers.Contains(ex.Number))
.CircuitBreakerAsync(
1,
TimeSpan.MaxValue,
(ex, _, ctx) => ctx.GetLogger()?.LogError(ex, "{@Operation} Encountered a Constraint Violation. Params:{@Param} Sql:{@Sql}", ctx.OperationKey, ctx[ParamContextKey], ctx[SqlContextKey]),
ctx => { }
);
var resiliencyPolicy = timeoutPolicy
.WrapAsync(transientPolicy)
.WrapAsync(networkPolicy)
.WrapAsync(constraintPolicy);
return resiliencyPolicy;
}
}
The Context
As you can see we make a lot of use of the Context in our policies. The Context object is scoped to a particular execution of a policy and provides a way to pass contextual information about the delegate that's being executed. Learn more with Using Execution Context in Polly We use it to store the requested sql script, the parameters, our logger and our connection. So to make this simpler we have a Context helper static class.
public static class ContextHelper
{
public static readonly string LoggerContextKey = nameof(LoggerContextKey);
public static readonly string SqlContextKey = nameof(SqlContextKey);
public static readonly string ParamContextKey = nameof(ParamContextKey);
public static readonly string ConnectionContextKey = nameof(ConnectionContextKey);
public static Context NewContext(
SqlConnection connection,
ILogger logger,
string sql,
object param,
string operationKey)
{
return new Context(operationKey, new Dictionary<string, object>()
{
{ ConnectionContextKey, connection },
{ LoggerContextKey, logger },
{ SqlContextKey, sql },
{ ParamContextKey, param }
});
}
public static ILogger GetLogger(this Context ctx)
=> ctx[LoggerContextKey] as ILogger;
public static bool TryGetConnection(this Context ctx, out SqlConnection connection)
=> (connection = ctx[ConnectionContextKey] as SqlConnection) is not null ? true : false;
}
Registration With DI
We need to register three items with our DI: A factory for our SqlConnection, our resiliency policy, and our yet to be revealed SqlDapperClient.
public delegate SqlConnection SqlConnectionFactory();
public static void AddSqlDapperClient(this IServiceCollection services, string connectionString)
{
services.AddSingleton<SqlConnectionFactory>(() => new SqlConnection(connectionString));
services.AddScoped(_ => SqlResiliencyPolicy.GetSqlResiliencyPolicy());
services.AddScoped<ISqlDapperClient, SqlDapperClient>();
}
Notice that our Resiliency Policy and Dapper Client are scoped. This means we'll get a new instance for each request and we're not sharing the policy across requests. We could have used a more elaborate setup if we needed our client to be singleton and used mutable Context to affect the state of our resiliency policy in the case of, for instance, the Constraint Violation exception, but we're just keeping it scoped here for simplicities sake.
The Dapper Wrapper, err... Client
Dapper is primarily a set of extension methods on IDbConnection. Now, while it is highly recommended to integration test your data access layer, I find it's also useful to unit test the layer above that, the layer that interacts with data access should be tested to ensure it passes the correct sql script and expected arguments. To facilitate this we need a way to mock the behavior of Dapper. Mocking the complete set of interactions from IDbConnection to a SqlCommand is laborious and we can simply create a client whose only purpose is to expose the extension methods of Dapper as a mockable interface and whose design is one that necessitate integration testing with a testing database. We expose an interface with the methods we'd like to mock, this is just a sampling of what Dapper offers but can easily be expanded.
public interface ISqlDapperClient
{
Task<int> ExecuteAsync(string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null);
Task<T> ExecuteScalarAsync<T>(string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null);
Task<IEnumerable<T>> QueryAsync<T>(string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null);
Task<T> QueryFirstOrDefaultAsync<T>(string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null);
Task<IEnumerable<TReturn>> QueryAsync<TFirst, TSecond, TReturn>(string sql, Func<TFirst, TSecond, TReturn> map, object param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null);
Task<IEnumerable<TReturn>> QueryAsync<TFirst, TSecond, TThird, TReturn>(string sql, Func<TFirst, TSecond, TThird, TReturn> map, object param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null);
}
Then in the implementation we inject our Logger, Resiliency Policy and our SqlConnection Factory and proxy the Dapper calls through our resiliency policy.
public class SqlDapperClient : ISqlDapperClient
{
private readonly ILogger<SqlDapperClient> logger;
private readonly SqlConnectionFactory connectionFactory;
private readonly IAsyncPolicy resiliencyPolicy;
public SqlDapperClient(
ILogger<SqlDapperClient> logger,
SqlConnectionFactory connectionFactory,
IAsyncPolicy resiliencyPolicy)
{
this.logger = logger ?? throw new ArgumentNullException(nameof(logger));
this.connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory));
this.resiliencyPolicy = resiliencyPolicy ?? throw new ArgumentNullException(nameof(resiliencyPolicy));
}
public Task<int> ExecuteAsync(string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null)
=> ExecuteWithResiliency((s, p, c) => c.ExecuteAsync(s, p, transaction, commandTimeout, commandType), sql, param);
public Task<T> ExecuteScalarAsync<T>(string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null)
=> ExecuteWithResiliency((s, p, c) => c.ExecuteScalarAsync<T>(s, p, transaction, commandTimeout, commandType), sql, param);
public Task<T> QueryFirstOrDefaultAsync<T>(string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null)
=> ExecuteWithResiliency((s, p, c) => c.QueryFirstOrDefaultAsync<T>(s, p, transaction, commandTimeout, commandType), sql, param);
public Task<IEnumerable<T>> QueryAsync<T>(string sql, object param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null)
=> ExecuteWithResiliency((s, p, c) => c.QueryAsync<T>(s, p, transaction, commandTimeout, commandType), sql, param);
public Task<IEnumerable<TReturn>> QueryAsync<TFirst, TSecond, TReturn>(string sql, Func<TFirst, TSecond, TReturn> map, object param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null)
=> ExecuteWithResiliency((s, p, c) => c.QueryAsync(s, map, p, transaction, buffered, splitOn, commandTimeout, commandType), sql, param);
public Task<IEnumerable<TReturn>> QueryAsync<TFirst, TSecond, TThird, TReturn>(string sql, Func<TFirst, TSecond, TThird, TReturn> map, object param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null)
=> ExecuteWithResiliency((s, p, c) => c.QueryAsync(s, map, p, transaction, buffered, splitOn, commandTimeout, commandType), sql, param);
private async Task<T> ExecuteWithResiliency<T>(Func<string, object, SqlConnection, Task<T>> connectionFunc, string sql, object param = null, [CallerMemberName] string operation = "")
{
using var connection = connectionFactory();
return await resiliencyPolicy.ExecuteAsync(
ctx => connectionFunc(sql, param, connection),
NewContext(connection, logger, sql, param, operation));
}
}
Notice we bring everything down to the ExecuteWithResiliency method that captures the extension method we want to execute, the sql we want to run and the parameters object that we pass to Dapper. In this method we create the connection and the instance of the execution Context and run our captured extension method with the provided arguments.
Summary
We have seen how to make a SqlDapperClient client resilient to transient errors with a global timeout on operations and reject constraint violations. And we've also seen how to make great use of Polly's execution Context to pass execution specific details to our policies. Finally we introduced a mockable wrapper for Dapper that will let us assert the against the passed arguments to the underlying extension methods. That's it for this post, hope you enjoyed it.
