diff --git a/docs/ActiveActive.md b/docs/ActiveActive.md index e6328910e..2bcc069a8 100644 --- a/docs/ActiveActive.md +++ b/docs/ActiveActive.md @@ -553,7 +553,24 @@ var value = await db.StringGetAsync("mykey"); > You can call `WithRetry` on any database (`IDatabase` or `IDatabaseAsync`), but the wrapper it returns exposes only the **async** API — there is no synchronous form, since retrying may inherently have delays. It cannot wrap a batch or an existing transaction, nor an already-retrying database. -A retrying database can still *create* a transaction: `retryDb.CreateTransaction()` returns an `ITransactionAsync` whose `ExecuteAsync` is retried as a single unit. Each attempt replays the queued operations (and any `WATCH` constraints) against a fresh `MULTI`/`EXEC` - and, in an Active:Active group, onto whichever member is active at the time - so a transaction can ride out a failover just like a single command; the per-operation tasks handed back at build time resolve from the winning attempt. The retry-category gate (below) still applies, using the *most* side-effecting operation in the transaction: a transaction containing an `INCR` is treated as `CommandRetryWriteAccumulating`, so the default policy will not retry it unless you raise `MaxCommandRetryCategory`. +> **`asyncState` is not respected.** A database's `asyncState` is stamped onto the task produced by a single dispatch, but a retrying database hands back its own task spanning however many attempts the operation takes, and the per-operation tasks from `retryDb.CreateTransaction()` are durable proxies that outlive any single attempt. Neither can carry it. Rather than dropping the state silently, both refuse it: `conn.GetDatabase(0, asyncState).WithRetry(policy)` throws `InvalidOperationException`, as does `retryDb.CreateTransaction(asyncState)`. Wrap a database obtained without an `asyncState` - note that a key-prefixed view of a state-carrying database is refused too, since it inherits the inner state. + +A retrying database can still *create* a transaction: `retryDb.CreateTransaction()` returns an `ITransactionAsync` whose `ExecuteAsync` is retried as a single unit. Each attempt replays the queued operations (and any `WATCH` constraints) against a fresh `MULTI`/`EXEC` - and, in an Active:Active group, onto whichever member is active at the time - so a transaction can ride out a failover just like a single command; the per-operation tasks handed back at build time resolve from the winning attempt. The retry-category gate (below) applies to the transaction as a whole, using the *most* side-effecting operation in it: a transaction containing an `INCR` counts as `CommandRetryWriteAccumulating`. In practice that gate rarely blocks a transaction, because the faults that a transaction is most likely to hit - a `MULTI`/`EXEC` the server *rejected* wholesale - are known not to have applied anything, and the category is not consulted in that case (see [Known-not-applied faults](#known-not-applied-faults)). + +#### Losing a `WATCH` race + +A transaction with conditions can fail to commit in two different ways, and `Execute`/`ExecuteAsync` reports `false` for both. `ITransaction.WasWatchConflict` (also on `ITransactionAsync`, so it works for retrying transactions too) is what tells them apart: + +- **a condition was not satisfied** - the transaction is abandoned electively, and no `EXEC` is ever sent. `WasWatchConflict` is `false`, and the offending `ConditionResult.WasSatisfied` is `false` too. Re-attempting is pointless: the value really was not what you asserted, so a replay would assert the same thing again and fail the same way. +- **another connection modified a watched key** - between this connection's conditions being evaluated and its `EXEC` arriving, some *other* client wrote to one of the keys being watched on behalf of those conditions, so the server refused the `EXEC`. `WasWatchConflict` is `true`, and `WasSatisfied` is `true` for every condition: your assertions were all correct, you just lost a race with a concurrent writer. This is contention, not a fault - nothing was applied and nothing is broken. + +Note that only *other connections* can cause this. Your own writes on this connection cannot: everything you queue inside the transaction is applied atomically by the `EXEC` itself, after the watch has already been satisfied. + +The second case is the one Redis's `WATCH`/`MULTI`/`EXEC` idiom expects you to retry, so a retrying transaction does exactly that, bounded by `MaxAttemptsOnWatchConflict` (default 3). Each re-attempt re-issues the constraints, so a transaction whose condition has genuinely stopped holding (because the concurrent writer left a value you were not expecting) converges on an ordinary elective abort rather than looping. Because it is contention rather than a fault it gets its own budget: `MaxAttempts` is untouched, `RetryDelay` is not applied (only `JitterMax`), no failover is attempted, and `MaxCommandRetryCategory` does not apply - nothing happened, so there is nothing to double-apply. Set `MaxAttemptsOnWatchConflict = 1` to restore the plain "report `false` and let me deal with it" behaviour. On a retrying transaction, `WasWatchConflict` describes the *final* attempt: `false` if it eventually committed, `true` if it ran out of attempts still losing the race. + +In either case every per-operation task is cancelled, so `await`ing one throws `OperationCanceledException` rather than hanging. + +How likely is this in practice? Much less so than in `redis-cli`-style usage, where a `WATCH` can sit open for as long as the application takes to think. SE.Redis buffers the whole transaction and sends the `WATCH`, the condition reads, the `MULTI`, the queued commands and the `EXEC` as a single dispatch on a multiplexed connection, so a competing writer has only the gap between the condition reads and the `EXEC` landing to squeeze into. Small - but not zero, and on a busy key it will happen. ### Configuring the retry policy @@ -584,6 +601,7 @@ The same works for a single connection via `ConfigurationOptions.RetryPolicy`. ` | `JitterMax` | 0.5 seconds | Upper bound of the additional random delay added to each retry, to avoid stampedes | | `FailoverDelay` | 5 seconds | Maximum time to wait for a failover, when a retry is gated on one happening | | `MaxCommandRetryCategory` | `CommandRetryWriteLastWins` | The most side-effecting command category that will be retried (see below) | +| `MaxAttemptsOnWatchConflict` | 3 | Attempts allowed for a *conditional transaction* that keeps losing a `WATCH` race; separate from `MaxAttempts`, and `1` disables re-attempting | ```csharp RetryPolicy policy = new RetryPolicy.Builder @@ -603,10 +621,23 @@ Retrying is not free of consequence: replaying `INCR` after an ambiguous failure For the built-in typed methods (`StringGet`, `StringSet`, `HashSet`, ...) the library assigns the appropriate category automatically, so retries "just work" within the default policy. +#### Known-not-applied faults + +The category prices the *ambiguity* of a replay, not the write itself. If we know the operation never took effect, re-issuing it is a first attempt rather than a repeat: it cannot double-apply anything, so the category is not consulted at all and even an `INCR` is retried under the default policy. Two things give us that certainty: + +- **the client never wrote it** - the message was still waiting to be sent, or sitting in the backlog, when the connection failed. +- **the server explicitly rejected it because of its own state** - `LOADING`, `CLUSTERDOWN`, `MASTERDOWN`, `TRYAGAIN`, `MOVED`, `ASK`, `READONLY`, `MISCONF`, `NOREPLICAS`, `BUSY`, `max clients`. The server can only report these *before* running anything. + +`FaultContext.NotApplied` exposes this to custom policies. It is deliberately conservative, and in particular a bare error reply is *not* enough on its own: a Lua script can fail part-way through having already written something, and it propagates the inner error (`WRONGTYPE`, `OOM`, ...) verbatim, so those kinds stay ambiguous. Timeouts, and connection loss after the request was sent, are always ambiguous - which is exactly where `MaxCommandRetryCategory` earns its keep. + +An explicit `CommandRetryNever` is still an absolute veto, as is an operation with no category at all (see below): certainty about *whether* it ran does not tell us that re-running it is meaningful. + ### Custom commands: `Execute` and `ScriptEvaluate` The library cannot infer the side-effects of a command it doesn't recognise — and that includes arbitrary commands issued via `Execute`/`ExecuteAsync`, and Lua run via `ScriptEvaluate`/`ScriptEvaluateAsync` (whose effect depends entirely on the script). Such commands are therefore treated **pessimistically**: an uncategorised command defaults to `CommandRetryNever` and is *not* retried. +Note that `Execute`/`ExecuteAsync` do try to *parse* the command name first, so `Execute("get", key)` is recognised as `GET` and picks up that command's category (read-only) automatically; only genuinely unrecognised command names fall back to `CommandRetryNever`. + The categories, from safest to most dangerous, are: | `CommandFlags` value | Meaning | diff --git a/src/StackExchange.Redis/Availability/DatabaseExtensions.cs b/src/StackExchange.Redis/Availability/DatabaseExtensions.cs index 646528884..c082259d1 100644 --- a/src/StackExchange.Redis/Availability/DatabaseExtensions.cs +++ b/src/StackExchange.Redis/Availability/DatabaseExtensions.cs @@ -13,13 +13,25 @@ public static class DatabaseExtensions /// SE.Redis concepts, so can respond to server failover events, apply circuit-breaker rules, and /// respect command effect categorization. /// - /// The database to wrap. + /// The database to wrap; this must not be a batch, a transaction, an + /// already-retrying database, or a database carrying an asyncState (see remarks). /// /// The policy to apply; when (the default), the policy configured for the /// underlying connection is used - for a connection group, /// for a single connection, else /// . /// + /// + /// asyncState is not supported. A database's asyncState is stamped onto the task + /// produced by a single dispatch, but a retrying database hands back its own task spanning however + /// many attempts the operation takes; the same is true of the per-operation tasks handed out by + /// on such a database. Rather than dropping + /// the state silently, both refuse it: wrapping a database obtained via + /// GetDatabase(db, asyncState) throws, as does supplying an asyncState when creating a + /// transaction from a retrying database. + /// + /// If is a batch, a + /// transaction, already retrying, or carries an asyncState. [Experimental(Experiments.ActiveActive, UrlFormat = Experiments.UrlFormat)] public static IDatabaseAsync WithRetry(this IDatabaseAsync database, RetryPolicy? retryPolicy = null) => new RetryDatabase(database, retryPolicy ?? ResolveRetryPolicy(database)); diff --git a/src/StackExchange.Redis/Availability/FaultContext.cs b/src/StackExchange.Redis/Availability/FaultContext.cs index 4321630af..0c70f5d94 100644 --- a/src/StackExchange.Redis/Availability/FaultContext.cs +++ b/src/StackExchange.Redis/Availability/FaultContext.cs @@ -27,6 +27,7 @@ public FaultContext(Exception fault) var kind = RedisErrorKind.None; _connectionFailureType = ConnectionFailureType.None; var flags = CommandFlags.None; + var status = CommandStatus.Unknown; switch (fault) { case RedisServerException server: @@ -37,16 +38,20 @@ public FaultContext(Exception fault) _connectionFailureType = connection.FailureType; kind = RedisErrorKind.ConnectionFault; flags = connection.Flags; + status = connection.CommandStatus; break; case RedisTimeoutException timeout: kind = RedisErrorKind.Timeout; flags = timeout.Flags; + status = timeout.Commandstatus; break; case TimeoutException: kind = RedisErrorKind.Timeout; break; } + NotApplied = IsKnownNotApplied(kind, status); + if (kind is not RedisErrorKind.None & _connectionFailureType is ConnectionFailureType.None) { // fill in some blanks @@ -93,8 +98,51 @@ public FaultContext(Exception fault) /// public RedisErrorKind ErrorKind { get; } + /// + /// Indicates that the operation is known *not* to have been applied by the server - either because it + /// never left the client, or because the server explicitly rejected it due to its own state (still + /// loading, cluster down, writes refused, and so on). Retrying such an operation is a *first* attempt + /// rather than a repeat, so it cannot double-apply a side-effect; therefore + /// ignores in this case. + /// + /// + /// This is deliberately conservative: it is only reported for conditions that the server can *only* + /// raise before running anything (so a Lua script that failed part-way through cannot be mistaken for + /// one that never ran), and for messages the client knows it never wrote. Everything else - notably + /// timeouts, and connection loss after the request was sent - remains ambiguous and is not flagged. + /// + public bool NotApplied { get; } + /// /// The connection failure type associated with the fault, if any. /// public ConnectionFailureType ConnectionFailureType => _connectionFailureType; + + private static bool IsKnownNotApplied(RedisErrorKind kind, CommandStatus status) + { + // the client never handed it to the socket, so the server cannot have seen it + if (status is CommandStatus.WaitingToBeSent or CommandStatus.WaitingInBacklog) return true; + + // an error *reply* usually means the server declined to run the command, but not always: a Lua + // script can fail part-way through, having already applied earlier writes, and it propagates the + // inner error verbatim (WRONGTYPE, and so on). So we only trust the conditions that describe the + // *server's own state*, which it can only report before running anything. + switch (kind) + { + case RedisErrorKind.Loading: // still loading the dataset + case RedisErrorKind.ClusterDown: // slot not currently served + case RedisErrorKind.MasterDown: // replica cannot serve, primary is unavailable + case RedisErrorKind.TryAgain: // slot mid-migration + case RedisErrorKind.Moved: // wrong node; this one did not run it + case RedisErrorKind.Ask: // ditto, mid-migration + case RedisErrorKind.ReadOnly: // writes refused by a replica + case RedisErrorKind.Misconfigured: // e.g. persistence failing, so writes are refused + case RedisErrorKind.NoReplicas: // not enough replicas to accept the write + case RedisErrorKind.Busy: // a script is hogging the server + case RedisErrorKind.MaxClients: // refused at the door + return true; + default: + return false; + } + } } diff --git a/src/StackExchange.Redis/Availability/RetryController.cs b/src/StackExchange.Redis/Availability/RetryController.cs index 72bbf8c33..a47d79bbc 100644 --- a/src/StackExchange.Redis/Availability/RetryController.cs +++ b/src/StackExchange.Redis/Availability/RetryController.cs @@ -13,7 +13,7 @@ namespace StackExchange.Redis.Availability; /// internal sealed class RetryController { - private readonly int _maxBeforeFailover, _maxAttempts, _delayMillis, _jitterMillis, _failoverMillis; + private readonly int _maxBeforeFailover, _maxAttempts, _delayMillis, _jitterMillis, _failoverMillis, _maxWatchAttempts; private readonly RetryPolicy _policy; public RetryController(RetryPolicy policy, DatabaseFeatureFlags features) @@ -29,9 +29,11 @@ public RetryController(RetryPolicy policy, DatabaseFeatureFlags features) _delayMillis = ToMilliseconds(policy.RetryDelay); _failoverMillis = ToMilliseconds(policy.FailoverDelay); _jitterMillis = ToMilliseconds(policy.JitterMax); + _maxWatchAttempts = policy.MaxAttemptsOnWatchConflict; Debug.Assert(_maxAttempts >= 1 && _maxBeforeFailover >= 1, "attempt counts should be validated by RetryPolicy"); Debug.Assert(_delayMillis >= 0 && _jitterMillis >= 0 && _failoverMillis >= 0, "delays should be validated by RetryPolicy"); + Debug.Assert(_maxWatchAttempts >= 1, "watch-conflict attempts should be validated by RetryPolicy"); static int ToMilliseconds(TimeSpan value) => (int)(value.Ticks / TimeSpan.TicksPerMillisecond); } @@ -41,6 +43,21 @@ public RetryController(RetryPolicy policy, DatabaseFeatureFlags features) /// public RetryPolicy Policy => _policy; + /// + /// How many times a conditional transaction may be attempted when the server keeps rejecting the + /// EXEC due to watch contention; see . + /// + public int MaxWatchConflictAttempts => _maxWatchAttempts; + + /// + /// The pause before re-attempting a transaction that lost a WATCH race. Contention, not a fault: + /// no backoff, just jitter to avoid two callers colliding again in lock-step. + /// + public Task WatchConflictDelayAsync() + => _jitterMillis is 0 + ? Task.CompletedTask + : Task.Delay(ServerSelectionStrategy.SharedRandom.Next(_jitterMillis), CancellationToken.None); + /// /// Whether it is ever worth capturing the next-failover token: only when there is more than one /// attempt and the failover threshold sits below the attempt cap. diff --git a/src/StackExchange.Redis/Availability/RetryDatabase.cs b/src/StackExchange.Redis/Availability/RetryDatabase.cs index 958299db2..ea037d786 100644 --- a/src/StackExchange.Redis/Availability/RetryDatabase.cs +++ b/src/StackExchange.Redis/Availability/RetryDatabase.cs @@ -16,6 +16,9 @@ internal partial class RetryDatabase : IDatabaseAsync, IInternalDatabaseAsync DatabaseFeatureFlags IInternalDatabaseAsync.GetFeatures(out string name) => _inner.GetFeatures(out name) | DatabaseFeatureFlags.Retry; + // never: we refuse to wrap a database that carries one (see Validate) + object? IInternalDatabaseAsync.AsyncState => null; + /// public override string ToString() => this.BuildString(); @@ -28,11 +31,25 @@ public CancellationToken GetNextFailover() => _controller.TracksFailover ? _inner.GetNextFailover() : CancellationToken.None; public RetryDatabase(IDatabaseAsync inner, RetryPolicy policy) - // cannot nest retry, and cannot issue retries *inside* a batch/transaction - : this(inner, policy, inner.RejectFlags(DatabaseFeatureFlags.Batch | DatabaseFeatureFlags.Transaction | DatabaseFeatureFlags.Retry)) + : this(inner, policy, Validate(inner)) { } + private static DatabaseFeatureFlags Validate(IDatabaseAsync inner) + { + // cannot nest retry, and cannot issue retries *inside* a batch/transaction + var features = inner.RejectFlags(DatabaseFeatureFlags.Batch | DatabaseFeatureFlags.Transaction | DatabaseFeatureFlags.Retry); + + // async-state is stamped onto the task that a *single* attempt produces; a retrying database hands + // back its own durable task that spans however many attempts it takes, so it cannot preserve the + // state. Refuse rather than dropping it silently. See also CreateTransaction, below. + if (inner.GetAsyncState() is not null) ThrowAsyncState(); + return features; + } + + internal static void ThrowAsyncState() => throw new InvalidOperationException( + "Retrying databases do not support asyncState; the tasks they hand back are not the tasks that were sent to the server."); + // test-only: supply the inner database's feature set directly (in particular whether failover is // available), instead of probing a live inner - so that failover behaviour can be exercised over a // null inner without a full IDatabaseAsync double. @@ -43,8 +60,14 @@ internal RetryDatabase(IDatabaseAsync inner, RetryPolicy policy, DatabaseFeature } ITransactionAsync IDatabaseAsync.CreateTransaction(object? asyncState) + { + // as per the constructor: the per-operation tasks handed out at build time are durable proxies + // that outlive any single attempt, so they cannot carry a per-attempt async-state + if (asyncState is not null) ThrowAsyncState(); + // the inner database creates the "real" (one-shot) transactions we replay against each attempt - => new RetryTransaction(_inner, _controller, asyncState); + return new RetryTransaction(_inner, _controller); + } public int Database => _inner.Database; diff --git a/src/StackExchange.Redis/Availability/RetryPolicy.cs b/src/StackExchange.Redis/Availability/RetryPolicy.cs index ffc2a3142..ada4dd891 100644 --- a/src/StackExchange.Redis/Availability/RetryPolicy.cs +++ b/src/StackExchange.Redis/Availability/RetryPolicy.cs @@ -17,6 +17,7 @@ public class RetryPolicy { internal const int DefaultMaxAttempts = 3; internal const int DefaultMaxAttemptsBeforeFailover = 1; + internal const int DefaultMaxAttemptsOnWatchConflict = 3; internal const CommandFlags DefaultMaxCommandRetryCategory = CommandFlags.CommandRetryWriteLastWins; internal static readonly TimeSpan DefaultRetryDelay = TimeSpan.FromSeconds(1); internal static readonly TimeSpan DefaultJitterMax = TimeSpan.FromMilliseconds(500); @@ -40,7 +41,7 @@ public class RetryPolicy /// - use to obtain the standard policy. /// protected RetryPolicy() - : this(DefaultMaxAttempts, DefaultMaxAttemptsBeforeFailover, DefaultRetryDelay, DefaultJitterMax, DefaultFailoverDelay, DefaultMaxCommandRetryCategory) + : this(DefaultMaxAttempts, DefaultMaxAttemptsBeforeFailover, DefaultMaxAttemptsOnWatchConflict, DefaultRetryDelay, DefaultJitterMax, DefaultFailoverDelay, DefaultMaxCommandRetryCategory) { } @@ -52,6 +53,7 @@ protected RetryPolicy(Builder builder) : this( Validate(builder).MaxAttempts, builder.MaxAttemptsBeforeFailover, + builder.MaxAttemptsOnWatchConflict, builder.RetryDelay, builder.JitterMax, builder.FailoverDelay, @@ -62,6 +64,7 @@ protected RetryPolicy(Builder builder) private RetryPolicy( int maxAttempts, int maxAttemptsBeforeFailover, + int maxAttemptsOnWatchConflict, TimeSpan retryDelay, TimeSpan jitterMax, TimeSpan failoverDelay, @@ -69,6 +72,7 @@ private RetryPolicy( { MaxAttempts = maxAttempts; MaxAttemptsBeforeFailover = maxAttemptsBeforeFailover; + MaxAttemptsOnWatchConflict = maxAttemptsOnWatchConflict; RetryDelay = retryDelay; JitterMax = jitterMax; FailoverDelay = failoverDelay; @@ -90,6 +94,23 @@ private RetryPolicy( /// public int MaxAttemptsBeforeFailover { get; } + /// + /// The maximum number of times a *conditional* transaction may be attempted when the only problem is + /// that the server rejected the EXEC because a watched key changed underneath it. Defaults to 3; + /// a value of 1 disables re-attempting such transactions. + /// + /// + /// This is deliberately separate from : a watch conflict is contention, + /// not a fault. Nothing was applied, nothing is broken, and the right response is to re-read the + /// conditions and try again immediately - so no is applied (only + /// ), no failover is attempted, and does + /// not apply. Each re-attempt re-issues the WATCH constraints, so a transaction whose condition + /// has genuinely stopped holding converges on an ordinary elective abort rather than looping. + /// Only transactions with conditions can be affected: without a condition there is no + /// WATCH, so there is nothing to conflict. + /// + public int MaxAttemptsOnWatchConflict { get; } + /// /// Gets the time to wait between retries that are *not* dependent on a failover happening. Defaults to 1 second. /// @@ -120,7 +141,7 @@ private RetryPolicy( public virtual RetryResult CanRetry(in FaultContext fault) { var actual = fault.Flags & Message.MaskRetryCategory; - if (actual is 0) actual = CommandFlags.CommandRetryWriteAccumulating; // if not set, assume similar to INCR + if (actual is 0) actual = CommandFlags.CommandRetryNever; // if not set, assume the worst (as FaultContext does) if (actual is CommandFlags.CommandRetryNever) { @@ -128,7 +149,10 @@ public virtual RetryResult CanRetry(in FaultContext fault) return RetryResult.None; } - if (actual > MaxCommandRetryCategory) // note this also covers CommandRetryAlways + // the category exists to price the *ambiguity* of a replay: if we know the operation never took + // effect, re-issuing it is a first attempt rather than a repeat, so it cannot double-apply and the + // side-effect scale is irrelevant. (CommandRetryNever above is still an absolute veto.) + if (actual > MaxCommandRetryCategory && !fault.NotApplied) { // side-effects are beyond what the policy allows return RetryResult.None; @@ -155,6 +179,11 @@ private static Builder Validate(Builder builder) // values < 1 can never be hit by the attempt counter (which starts at 1), so they would *silently* // disable failover rather than erroring if (builder.MaxAttemptsBeforeFailover < 1) throw new ArgumentOutOfRangeException(nameof(builder.MaxAttemptsBeforeFailover), builder.MaxAttemptsBeforeFailover, "At least one attempt is required before failover."); + + // this one counts *attempts*, so 1 means "try once, do not re-attempt"; zero or negative is + // meaningless rather than a way to say "never execute" + if (builder.MaxAttemptsOnWatchConflict < 1) throw new ArgumentOutOfRangeException(nameof(builder.MaxAttemptsOnWatchConflict), builder.MaxAttemptsOnWatchConflict, "At least one attempt is required; use 1 to disable re-attempting on watch conflicts."); + if (builder.RetryDelay < TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(builder.RetryDelay), builder.RetryDelay, "A non-negative retry delay is required."); if (builder.JitterMax < TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(builder.JitterMax), builder.JitterMax, "A non-negative jitter bound is required."); if (builder.FailoverDelay < TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(builder.FailoverDelay), builder.FailoverDelay, "A non-negative failover delay is required."); @@ -195,6 +224,7 @@ public Builder(RetryPolicy policy) { MaxAttempts = policy.MaxAttempts; MaxAttemptsBeforeFailover = policy.MaxAttemptsBeforeFailover; + MaxAttemptsOnWatchConflict = policy.MaxAttemptsOnWatchConflict; RetryDelay = policy.RetryDelay; JitterMax = policy.JitterMax; FailoverDelay = policy.FailoverDelay; @@ -211,6 +241,9 @@ public Builder(RetryPolicy policy) /// public int MaxAttemptsBeforeFailover { get; set; } = DefaultMaxAttemptsBeforeFailover; + /// + public int MaxAttemptsOnWatchConflict { get; set; } = DefaultMaxAttemptsOnWatchConflict; + /// /// The time to wait between retries that are *not* dependent on a failover happening. /// @@ -241,6 +274,7 @@ public RetryPolicy Create() // prefer the shared default instance when nothing has been customized if (MaxAttempts == DefaultMaxAttempts && MaxAttemptsBeforeFailover == DefaultMaxAttemptsBeforeFailover + && MaxAttemptsOnWatchConflict == DefaultMaxAttemptsOnWatchConflict && RetryDelay == DefaultRetryDelay && JitterMax == DefaultJitterMax && FailoverDelay == DefaultFailoverDelay @@ -249,7 +283,7 @@ public RetryPolicy Create() return DefaultInstance; } - return new RetryPolicy(MaxAttempts, MaxAttemptsBeforeFailover, RetryDelay, JitterMax, FailoverDelay, MaxCommandRetryCategory); + return new RetryPolicy(MaxAttempts, MaxAttemptsBeforeFailover, MaxAttemptsOnWatchConflict, RetryDelay, JitterMax, FailoverDelay, MaxCommandRetryCategory); } /// @@ -261,7 +295,12 @@ public RetryPolicy Create() private sealed class NoRetryPolicy : RetryPolicy { public static readonly NoRetryPolicy Instance = new(); - private NoRetryPolicy() { } + + // "never retries anything" has to include watch contention: that path does not consult CanRetry + // (nothing was applied, so there is no fault to judge), it is bounded purely by the attempt count + private NoRetryPolicy() : base(new Builder { MaxAttemptsOnWatchConflict = 1 }) + { + } public override RetryResult CanRetry(in FaultContext fault) => RetryResult.None; } diff --git a/src/StackExchange.Redis/Availability/RetryTransaction.cs b/src/StackExchange.Redis/Availability/RetryTransaction.cs index c79507bd1..0d87cd47d 100644 --- a/src/StackExchange.Redis/Availability/RetryTransaction.cs +++ b/src/StackExchange.Redis/Availability/RetryTransaction.cs @@ -19,18 +19,22 @@ internal sealed partial class RetryTransaction : IDatabaseAsync, ITransactionAsy { // Note: async-only, exactly like RetryDatabase - retrying is inherently delay-ish. private readonly IDatabaseAsync _source; - private readonly object? _asyncState; private readonly RetryController _controller; private readonly List _ops = new(); private List? _conditions; private int _executed; + private volatile bool _watchConflict; - public RetryTransaction(IDatabaseAsync source, RetryController controller, object? asyncState) + /// + // reports the *final* attempt's outcome: false if we eventually committed (or aborted electively), + // true if we ran out of watch-conflict attempts still losing the race + public bool WasWatchConflict => _watchConflict; + + public RetryTransaction(IDatabaseAsync source, RetryController controller) { _source = source; _controller = controller; - _asyncState = asyncState; } public int Database => _source.Database; @@ -83,9 +87,14 @@ public async Task ExecuteAsync(CommandFlags flags = CommandFlags.None) CancellationToken failover = _controller.TracksFailover ? _source.GetNextFailover() : CancellationToken.None; var conditions = _conditions; + + // watch contention gets its own budget; only meaningful when there are conditions to WATCH + int watchAttempt = 0; + int maxWatchAttempts = conditions is null ? 1 : _controller.MaxWatchConflictAttempts; while (true) { - var inner = _source.CreateTransaction(_asyncState); + // no async-state: RetryDatabase refuses one (the durable proxies below cannot carry it) + var inner = _source.CreateTransaction(); // replay the recorded constraints and operations onto this fresh, one-shot transaction; the // per-attempt tasks they return are forwarded to the durable proxies only on a clean execution @@ -103,9 +112,25 @@ public async Task ExecuteAsync(CommandFlags flags = CommandFlags.None) try { bool committed = await inner.ExecuteAsync(effectiveFlags).ConfigureAwait(false); + _watchConflict = inner.WasWatchConflict; // surfaced to our own caller, per attempt + + // The server rejected an EXEC we really did issue, because another connection changed a + // watched key: the conditions still held, nothing was applied, and we simply lost a race. + // Re-read and try again (which re-issues the WATCH constraints, so a condition that has + // genuinely stopped holding converges on an elective abort instead of looping). This is + // contention rather than a fault, so it neither consumes the fault budget nor waits for a + // failover, and the side-effect category does not apply - nothing happened. + if (!committed + && _watchConflict + && ++watchAttempt < maxWatchAttempts) + { + foreach (var op in _ops) op.Observe(); + await _controller.WatchConflictDelayAsync().ConfigureAwait(false); + continue; + } - // clean completion (committed, or electively aborted via a failed WATCH); forward the - // per-attempt outcomes onto the durable proxies and we're done + // clean completion (committed, or aborted); forward the per-attempt outcomes onto the + // durable proxies and we're done if (conditions is not null) { foreach (var c in conditions) c.ForwardSuccess(); diff --git a/src/StackExchange.Redis/ExceptionFactory.cs b/src/StackExchange.Redis/ExceptionFactory.cs index 8428d41af..0f6ab3cb5 100644 --- a/src/StackExchange.Redis/ExceptionFactory.cs +++ b/src/StackExchange.Redis/ExceptionFactory.cs @@ -1,4 +1,5 @@ using System; +using System.Collections; using System.Collections.Generic; using System.Security.Authentication; using System.Text; @@ -164,6 +165,37 @@ internal static Exception NoConnectionAvailable( return ex; } + /// + /// A connection failure produces a single exception describing the *connection*, which is then handed + /// to every message that was in flight. Sharing one instance across many unrelated callers is dubious + /// in itself ( is mutable, so each caller sees the others' additions), and + /// it discards the per-message detail the retry machinery needs: the command's retry category, and + /// whether this particular message had actually been written. Give each message its own. + /// + internal static Exception PerMessage(Exception shared, Message message) + { + // only connection failures get shared like this; anything else already describes one operation + if (shared is not RedisConnectionException conn + || (conn.Flags == message.Flags && conn.CommandStatus == message.Status)) + { + return shared; + } + + var ex = new RedisConnectionException( + conn.FailureType, + message.Flags, + conn.Message, + conn.InnerException, + message.Status); + foreach (DictionaryEntry entry in conn.Data) + { + ex.Data[entry.Key] = entry.Value; + } + ex.Data[DataSentStatusKey] = message.Status; // ...and correct this one for *this* message + if (conn.HelpLink is not null) ex.HelpLink = conn.HelpLink; + return ex; + } + internal static Exception? PopulateInnerExceptions(ReadOnlySpan serverSnapshot) { var innerExceptions = new List(); diff --git a/src/StackExchange.Redis/Interfaces/IInternalDatabaseAsync.cs b/src/StackExchange.Redis/Interfaces/IInternalDatabaseAsync.cs index 6abc0505a..6aea5db57 100644 --- a/src/StackExchange.Redis/Interfaces/IInternalDatabaseAsync.cs +++ b/src/StackExchange.Redis/Interfaces/IInternalDatabaseAsync.cs @@ -21,6 +21,13 @@ internal interface IInternalDatabaseAsync : IDatabaseAsync { DatabaseFeatureFlags GetFeatures(out string name); CancellationToken GetNextFailover(); + + /// + /// The async-state that this database stamps onto the tasks it hands out, if any; wrappers that + /// cannot preserve it (see RetryDatabase) need to know when one is present rather than + /// dropping it silently. + /// + object? AsyncState { get; } } /// @@ -67,6 +74,9 @@ internal static DatabaseFeatureFlags RejectFlags(this IDatabaseAsync database, D $"This operation is not compatible with feature(s): {overlap}"); } + internal static object? GetAsyncState(this IDatabaseAsync database) + => database is IInternalDatabaseAsync ida ? ida.AsyncState : null; + internal static CancellationToken GetNextFailover(this IDatabaseAsync database) { // get a CT that represents the next failover; you might be asking "shouldn't that be a Task getter?" - no, diff --git a/src/StackExchange.Redis/Interfaces/ITransactionAsync.cs b/src/StackExchange.Redis/Interfaces/ITransactionAsync.cs index a9c840dc1..14548c7d7 100644 --- a/src/StackExchange.Redis/Interfaces/ITransactionAsync.cs +++ b/src/StackExchange.Redis/Interfaces/ITransactionAsync.cs @@ -26,4 +26,28 @@ public interface ITransactionAsync : IDatabaseAsync /// /// The command flags to use. Task ExecuteAsync(CommandFlags flags = CommandFlags.None); + + /// + /// Whether the transaction failed to commit because the *server* rejected the EXEC: every + /// condition held, so MULTI/EXEC really was issued, but a key being watched on behalf of + /// those conditions was modified by another connection in the meantime. + /// + /// + /// A transaction that does not commit reports false from Execute for two quite + /// different reasons, and this is what tells them apart: + /// + /// A condition was not satisfied, so the transaction was abandoned without ever + /// issuing an EXEC. The value genuinely was not what was asserted, and re-running would assert + /// the same thing again. This property is false, and the offending + /// is also false. + /// A watched key was changed by a different connection between the + /// conditions being evaluated and the EXEC arriving. This property is true, every + /// is true, and re-reading and trying again is the + /// expected response - this is what the WATCH/MULTI/EXEC idiom is for. + /// + /// Either way nothing was applied, and every queued operation's task is cancelled. This can only + /// be true for a transaction that has conditions (without one there is no WATCH, so there + /// is nothing to conflict over), and is false before the transaction has been executed. + /// + bool WasWatchConflict { get; } } diff --git a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs index 306143f9a..c889a2c57 100644 --- a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs +++ b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs @@ -35,6 +35,9 @@ DatabaseFeatureFlags IInternalDatabaseAsync.GetFeatures(out string name) CancellationToken IInternalDatabaseAsync.GetNextFailover() => Inner.GetNextFailover(); + // this wrapper does not stamp its own async-state; it inherits whatever the inner database uses + object? IInternalDatabaseAsync.AsyncState => Inner.GetAsyncState(); + // the flags contributed by this wrapper itself (on top of the inner database); the batch and // transaction subclasses override to fold in their own flag, mirroring RedisDatabase/RedisBatch/ // RedisTransaction rather than relying on the inner instance to carry it. diff --git a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedTransaction.cs b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedTransaction.cs index 78e1a133a..3658d65ee 100644 --- a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedTransaction.cs +++ b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedTransaction.cs @@ -15,6 +15,9 @@ private protected override DatabaseFeatureFlags GetDatabaseFeatures() CommandFlags IInternalTransaction.GetAggregateRetryCategory() => Inner is IInternalTransaction it ? it.GetAggregateRetryCategory() : CommandFlags.CommandRetryNever; + /// + public bool WasWatchConflict => Inner.WasWatchConflict; + public ConditionResult AddCondition(Condition condition) => Inner.AddCondition(condition.MapKeys(GetMapFunction())); public bool Execute(CommandFlags flags = CommandFlags.None) => Inner.Execute(flags); diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 5bfa67ff0..049824536 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -515,8 +515,12 @@ private void AbandonPendingBacklog(Exception ex, PhysicalConnection? connection) { while (BacklogTryDequeue(out Message? next)) { - Multiplexer.OnMessageFaulted(next, ex); - next.SetExceptionAndComplete(ex, connection); + // as in PhysicalConnection.RecordMessageFailed: don't hand the same exception to every + // message. A backlogged message has provably never been written, which is exactly what the + // retry machinery needs to know, and it is lost if we share the connection-level instance. + var perMessage = ExceptionFactory.PerMessage(ex, next); + Multiplexer.OnMessageFaulted(next, perMessage); + next.SetExceptionAndComplete(perMessage, connection); } } diff --git a/src/StackExchange.Redis/PhysicalConnection.cs b/src/StackExchange.Redis/PhysicalConnection.cs index 3bfb8faba..80ef8621b 100644 --- a/src/StackExchange.Redis/PhysicalConnection.cs +++ b/src/StackExchange.Redis/PhysicalConnection.cs @@ -542,6 +542,10 @@ private void RecordMessageFailed(Message next, Exception? ex, string? origin, Ph } else { + // the connection-level exception is shared across every message being failed here; give this + // one its own, carrying *its* flags and sent-status so retry policy can reason about it + if (ex is not null) ex = ExceptionFactory.PerMessage(ex, next); + var bridge = connection?.BridgeCouldBeNull; if (bridge is not null) { diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 1e48387d6..427d8c4ad 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -27,6 +27,7 @@ StackExchange.Redis.IDatabaseAsync.Database.get -> int [SER007]StackExchange.Redis.Availability.RetryPolicy.JitterMax.get -> System.TimeSpan [SER007]StackExchange.Redis.Availability.RetryPolicy.MaxAttempts.get -> int [SER007]StackExchange.Redis.Availability.RetryPolicy.MaxAttemptsBeforeFailover.get -> int +[SER007]StackExchange.Redis.Availability.RetryPolicy.MaxAttemptsOnWatchConflict.get -> int [SER007]StackExchange.Redis.Availability.RetryPolicy.MaxCommandRetryCategory.get -> StackExchange.Redis.CommandFlags [SER007]StackExchange.Redis.Availability.RetryPolicy.RetryDelay.get -> System.TimeSpan [SER007]StackExchange.Redis.Availability.RetryPolicy.RetryPolicy() -> void @@ -37,6 +38,7 @@ StackExchange.Redis.Availability.DatabaseExtensions StackExchange.Redis.ITransactionAsync StackExchange.Redis.ITransactionAsync.AddCondition(StackExchange.Redis.Condition! condition) -> StackExchange.Redis.ConditionResult! StackExchange.Redis.ITransactionAsync.ExecuteAsync(StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +StackExchange.Redis.ITransactionAsync.WasWatchConflict.get -> bool [SER007]StackExchange.Redis.Availability.CircuitBreaker [SER007]StackExchange.Redis.Availability.CircuitBreaker.Accumulator [SER007]StackExchange.Redis.Availability.CircuitBreaker.Accumulator.Accumulator() -> void @@ -114,6 +116,7 @@ StackExchange.Redis.ITransactionAsync.ExecuteAsync(StackExchange.Redis.CommandFl [SER007]StackExchange.Redis.Availability.FaultContext.FaultContext(System.Exception! fault) -> void [SER007]StackExchange.Redis.Availability.FaultContext.Flags.get -> StackExchange.Redis.CommandFlags [SER007]StackExchange.Redis.Availability.FaultContext.IsFault.get -> bool +[SER007]StackExchange.Redis.Availability.FaultContext.NotApplied.get -> bool [SER007]StackExchange.Redis.RedisErrorKind.Ask = 15 -> StackExchange.Redis.RedisErrorKind [SER007]StackExchange.Redis.RedisErrorKind.Busy = 12 -> StackExchange.Redis.RedisErrorKind [SER007]StackExchange.Redis.RedisErrorKind.ClusterDown = 6 -> StackExchange.Redis.RedisErrorKind @@ -239,6 +242,8 @@ StackExchange.Redis.RedisTimeoutException.RedisTimeoutException(StackExchange.Re [SER007]StackExchange.Redis.Availability.RetryPolicy.Builder.JitterMax.set -> void [SER007]StackExchange.Redis.Availability.RetryPolicy.Builder.MaxAttemptsBeforeFailover.get -> int [SER007]StackExchange.Redis.Availability.RetryPolicy.Builder.MaxAttemptsBeforeFailover.set -> void +[SER007]StackExchange.Redis.Availability.RetryPolicy.Builder.MaxAttemptsOnWatchConflict.get -> int +[SER007]StackExchange.Redis.Availability.RetryPolicy.Builder.MaxAttemptsOnWatchConflict.set -> void [SER007]StackExchange.Redis.Availability.RetryPolicy.Builder.MaxAttempts.get -> int [SER007]StackExchange.Redis.Availability.RetryPolicy.Builder.MaxAttempts.set -> void [SER007]StackExchange.Redis.Availability.RetryPolicy.Builder.MaxCommandRetryCategory.get -> StackExchange.Redis.CommandFlags diff --git a/src/StackExchange.Redis/RedisTransaction.cs b/src/StackExchange.Redis/RedisTransaction.cs index 0fc790440..b23ade3d5 100644 --- a/src/StackExchange.Redis/RedisTransaction.cs +++ b/src/StackExchange.Redis/RedisTransaction.cs @@ -13,8 +13,13 @@ internal sealed class RedisTransaction : RedisDatabase, ITransaction, IInternalT { private List? _conditions; private List? _pending; + private TransactionMessage? _lastMessage; private object SyncLock => this; + /// + // set by TransactionProcessor when the server answers EXEC with a null array + public bool WasWatchConflict => _lastMessage?.WasWatchConflict == true; + // combine the retry categories of all queued operations, taking the most side-effecting (numerically // highest) - this is what a replay of the whole transaction would do. WATCH constraints are *not* // included: they live in _conditions, not _pending, and re-issuing them is what makes replay safe. @@ -195,7 +200,7 @@ private void QueueMessage(Message message) } processor = TransactionProcessor.Default; - return new TransactionMessage(Database, flags, cond, work); + return _lastMessage = new TransactionMessage(Database, flags, cond, work); } private sealed class QueuedMessage : Message @@ -279,6 +284,16 @@ internal override void SetExceptionAndComplete(Exception exception, PhysicalConn public bool IsAborted => command != RedisCommand.EXEC; + // the server rejected an EXEC we really did issue, because a watched key moved; volatile + // because it is written on the read loop and observed by whoever awaited the transaction + public bool WasWatchConflict + { + get => Volatile.Read(ref _watchConflict); + set => Volatile.Write(ref _watchConflict, value); + } + + private bool _watchConflict; + public override void AppendStormLog(StringBuilder sb) { base.AppendStormLog(sb); @@ -528,16 +543,23 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes if (reader.IsNull) // EXEC returned with a NULL { - if (tran.IsAborted) + // the server refused to apply the transaction because a watched key changed + // ("WATCH drift"); nothing was applied, so every queued operation must be + // brought to a terminal state - otherwise the caller's tasks hang forever. + // (in the electively-aborted case they were already cancelled in GetMessages; + // Cancel/Complete are idempotent, so doing it again is harmless) + muxer?.OnTransactionLog("Aborting wrapped messages (failed watch)"); + connection.Trace("Server aborted due to failed WATCH"); + + // distinguish "the server refused an EXEC we issued" from "we chose not to issue + // one"; only the former is worth re-attempting (see IInternalTransaction) + tran.WasWatchConflict = !tran.IsAborted; + + foreach (var op in wrapped) { - muxer?.OnTransactionLog("Aborting wrapped messages (failed watch)"); - connection.Trace("Server aborted due to failed WATCH"); - foreach (var op in wrapped) - { - var inner = op.Wrapped; - inner.Cancel(); - inner.Complete(connection); - } + var inner = op.Wrapped; + inner.Cancel(); + inner.Complete(connection); } SetResult(message, false); return true; diff --git a/tests/StackExchange.Redis.Tests/InProcessTestServer.cs b/tests/StackExchange.Redis.Tests/InProcessTestServer.cs index 2b14692be..2f6833ace 100644 --- a/tests/StackExchange.Redis.Tests/InProcessTestServer.cs +++ b/tests/StackExchange.Redis.Tests/InProcessTestServer.cs @@ -403,13 +403,15 @@ protected override void Dispose(bool disposing) if (disposing) _server.Dispose(); } */ - public void SetLatency(TimeSpan latency) => _latency = latency; + // written by the test thread, read by the server's own thread(s): store as ticks so the read/write can + // be made explicitly visible (a plain TimeSpan field can be missed indefinitely) + public void SetLatency(TimeSpan latency) => Volatile.Write(ref _latencyTicks, latency.Ticks); - private TimeSpan _latency = TimeSpan.Zero; + private long _latencyTicks; protected override ValueTask ClientPauseAsync(RedisClient client, in RedisRequest request) { - var latency = _latency; + var latency = TimeSpan.FromTicks(Volatile.Read(ref _latencyTicks)); if (latency > TimeSpan.Zero & request.KnownCommand != RedisCommand.QUIT) { Log($"[{client}] holding {request.Command} response by {latency.TotalMilliseconds}ms"); diff --git a/tests/StackExchange.Redis.Tests/RetryTests/CommandRetryPolicyUnitTests.cs b/tests/StackExchange.Redis.Tests/RetryTests/CommandRetryPolicyUnitTests.cs index 45720df2e..c1d6e1d51 100644 --- a/tests/StackExchange.Redis.Tests/RetryTests/CommandRetryPolicyUnitTests.cs +++ b/tests/StackExchange.Redis.Tests/RetryTests/CommandRetryPolicyUnitTests.cs @@ -18,10 +18,18 @@ private static RetryResult CanRetry(RedisErrorKind kind, CommandFlags flags, Ret return (policy ?? RetryPolicy.Default).CanRetry(in fault); } + // As above, but for an *ambiguous* fault: a timeout on a request we know was sent. We have no idea + // whether the server applied it, so this is the case the retry-category exists to price. + private static RetryResult CanRetryAmbiguous(CommandFlags flags, RetryPolicy? policy = null) + { + var fault = new FaultContext(new RedisTimeoutException(flags, "timeout", CommandStatus.Sent)); + return (policy ?? RetryPolicy.Default).CanRetry(in fault); + } + // The command's retry-category is checked against the policy's max category: the default max is // CommandRetryWriteLastWins, so anything at-or-below that is in-range, and anything with more - // side-effects is not. Using a transient LOADING fault so the error-kind check permits a retry - // whenever the category is in-range - isolating the category logic. + // side-effects is not. Using an ambiguous (timeout-after-send) fault, since that is where the + // category actually bites - see CanRetry_NotAppliedBypassesCategory for the other half. [Theory] [InlineData(CommandFlags.CommandRetryAlways, true)] [InlineData(CommandFlags.CommandRetryConnection, true)] @@ -31,13 +39,51 @@ private static RetryResult CanRetry(RedisErrorKind kind, CommandFlags flags, Ret [InlineData(CommandFlags.CommandRetryWriteAccumulating, false)] // beyond default max [InlineData(CommandFlags.CommandRetryServerAdmin, false)] [InlineData(CommandFlags.CommandRetryNever, false)] - [InlineData(CommandFlags.None, false)] // unspecified => assume worst (accumulating) => beyond default max + [InlineData(CommandFlags.None, false)] // unspecified => assume the worst => not retried public void CanRetry_CategoryVersusDefaultMax(CommandFlags category, bool expectRetry) { + var result = CanRetryAmbiguous(category); + Assert.Equal(expectRetry, result != RetryResult.None); + } + + // When the fault proves the operation never took effect (here: the server was still LOADING, so it + // rejected the command outright), a replay is a first attempt rather than a repeat - it cannot + // double-apply anything, so the side-effect category is irrelevant and the cap is bypassed. An + // explicit CommandRetryNever is still an absolute veto, as is an unspecified category. + [Theory] + [InlineData(CommandFlags.CommandRetryWriteAccumulating, true)] // beyond the default cap, but safe here + [InlineData(CommandFlags.CommandRetryServerAdmin, true)] + [InlineData(CommandFlags.CommandRetryNever, false)] // never means never + [InlineData(CommandFlags.None, false)] // we don't know what it is; don't guess + public void CanRetry_NotAppliedBypassesCategory(CommandFlags category, bool expectRetry) + { + // sanity: the same category against an ambiguous fault of the same "retryability" is refused + if (expectRetry) Assert.Equal(RetryResult.None, CanRetryAmbiguous(category)); + var result = CanRetry(RedisErrorKind.Loading, category); Assert.Equal(expectRetry, result != RetryResult.None); } + // The other source of certainty: the client knows it never wrote the message. Same conclusion, even + // though the fault itself (a connection failure) is otherwise ambiguous. + [Theory] + [InlineData(CommandStatus.WaitingToBeSent, true)] + [InlineData(CommandStatus.WaitingInBacklog, true)] + [InlineData(CommandStatus.Sent, false)] // may or may not have been applied + [InlineData(CommandStatus.Unknown, false)] + public void CanRetry_UnsentMessageBypassesCategory(CommandStatus status, bool expectRetry) + { + var fault = new FaultContext(new RedisConnectionException( + ConnectionFailureType.SocketFailure, + CommandFlags.CommandRetryWriteAccumulating, // beyond the default cap + "boom", + innerException: null, + commandStatus: status)); + + Assert.Equal(expectRetry, fault.NotApplied); + Assert.Equal(expectRetry, RetryPolicy.Default.CanRetry(in fault) != RetryResult.None); + } + // With an in-range category (== default max), the outcome is decided purely by whether the error // is transient: LOADING is worth retrying, WRONGTYPE is an application error that will not fix itself. [Theory] @@ -86,8 +132,8 @@ public void CanRetry_ServerSpecificRestrictsToSameServer(CommandFlags extra, Ret [InlineData(CommandFlags.CommandRetryServerAdmin, false)] // beyond default max public void CanRetry_ServerSpecificDoesNotAffectRange(CommandFlags category, bool expectRetry) { - var withoutFlag = CanRetry(RedisErrorKind.Loading, category); - var withFlag = CanRetry(RedisErrorKind.Loading, category | Message.CommandServerSpecific); + var withoutFlag = CanRetryAmbiguous(category); + var withFlag = CanRetryAmbiguous(category | Message.CommandServerSpecific); Assert.Equal(expectRetry, withoutFlag != RetryResult.None); Assert.Equal(expectRetry, withFlag != RetryResult.None); diff --git a/tests/StackExchange.Redis.Tests/RetryTests/ConnectionFaultDetailTests.cs b/tests/StackExchange.Redis.Tests/RetryTests/ConnectionFaultDetailTests.cs new file mode 100644 index 000000000..95781c82c --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RetryTests/ConnectionFaultDetailTests.cs @@ -0,0 +1,153 @@ +using StackExchange.Redis.Availability; +using Xunit; + +namespace StackExchange.Redis.Tests.RetryTests; + +/// +/// When a connection dies, one exception is built to describe the *connection*, and it used to be handed +/// verbatim to every message that was in flight. Sharing one exception instance across unrelated callers is +/// dubious in itself (Exception.Data is mutable), and it discards the per-message facts the retry +/// machinery needs: the command's retry category, and whether this particular message had actually been +/// written. Without them nothing at all is retryable after a connection failure, not even a plain +/// GET. is what splits them apart. +/// +public class ConnectionFaultDetailTests +{ + private static RedisConnectionException SharedConnectionFault() + { + // as built by PhysicalConnection.RecordConnectionFailed: describes the connection, knows nothing + // about any individual message + var ex = new RedisConnectionException( + ConnectionFailureType.SocketClosed, + CommandFlags.None, + "SocketClosed on 127.0.0.1:6379/Interactive"); + ex.Data["Redis-Version"] = "1.2.3"; + ex.Data["Redis-Server"] = "127.0.0.1:6379"; + return ex; + } + + private static Message Read() => Message.Create(0, CommandFlags.None, RedisCommand.GET, (RedisKey)"key"); + + private static Message AccumulatingWrite() => Message.Create(0, CommandFlags.None, RedisCommand.INCR, (RedisKey)"key"); + + // A message that was written before the socket died: the outcome is genuinely ambiguous, so the sent + // status must survive as-is, but the command's category has to come through - otherwise the policy sees + // "no category" and refuses to retry even a pure read. + [Fact] + public void SentMessage_CarriesCategoryAndSentStatus() + { + var shared = SharedConnectionFault(); + var message = Read(); + message.SetRequestSent(); + + var per = Assert.IsType(ExceptionFactory.PerMessage(shared, message)); + + Assert.NotSame(shared, per); + Assert.Equal(ConnectionFailureType.SocketClosed, per.FailureType); + Assert.Equal(CommandStatus.Sent, per.CommandStatus); + Assert.Equal(CommandFlags.CommandRetryReadOnly, per.Flags & Message.MaskRetryCategory); + + var ctx = new FaultContext(per); + Assert.False(ctx.NotApplied); // it was on the wire; we cannot know whether the server ran it + Assert.NotEqual(RetryResult.None, RetryPolicy.Default.CanRetry(in ctx)); // ...but a read is safe + + // for contrast: the shared exception the message used to receive is retryable for nothing at all + var sharedCtx = new FaultContext(shared); + Assert.Equal(RetryResult.None, RetryPolicy.Default.CanRetry(in sharedCtx)); + } + + // Same situation, accumulating write: the category comes through and correctly *blocks* the retry, since + // a replay could double-apply. The caller can still opt in by raising the cap. + [Fact] + public void SentAccumulatingWrite_RemainsGatedByCategory() + { + var message = AccumulatingWrite(); + message.SetRequestSent(); + + var per = Assert.IsType(ExceptionFactory.PerMessage(SharedConnectionFault(), message)); + Assert.Equal(CommandFlags.CommandRetryWriteAccumulating, per.Flags & Message.MaskRetryCategory); + + var ctx = new FaultContext(per); + Assert.False(ctx.NotApplied); + Assert.Equal(RetryResult.None, RetryPolicy.Default.CanRetry(in ctx)); + + RetryPolicy permissive = new RetryPolicy.Builder { MaxCommandRetryCategory = CommandFlags.CommandRetryWriteAccumulating }; + Assert.NotEqual(RetryResult.None, permissive.CanRetry(in ctx)); + } + + // A message that never left the client - still waiting to be written, or sitting in the backlog - is + // *provably* unapplied, which is the one case where even an accumulating write can be safely re-issued. + // That fact lives on the message, so sharing the connection's exception threw it away. + [Theory] + [InlineData(false)] // never handed to the bridge + [InlineData(true)] // queued in the backlog awaiting a healthy connection + public void UnsentMessage_IsKnownNotApplied(bool backlogged) + { + var message = AccumulatingWrite(); + if (backlogged) message.SetBacklogged(); + + var per = Assert.IsType(ExceptionFactory.PerMessage(SharedConnectionFault(), message)); + + Assert.Equal(backlogged ? CommandStatus.WaitingInBacklog : CommandStatus.WaitingToBeSent, per.CommandStatus); + + var ctx = new FaultContext(per); + Assert.True(ctx.NotApplied); + // accumulating, i.e. beyond the default cap - but nothing was applied, so there is nothing to repeat + Assert.NotEqual(RetryResult.None, RetryPolicy.Default.CanRetry(in ctx)); + } + + // The connection-level diagnostics are the useful part of these exceptions, so they have to come across; + // but the dictionaries must be independent, or one caller's annotations show up on another's exception. + [Fact] + public void SharedDiagnosticsAreCopied_ButNotShared() + { + var shared = SharedConnectionFault(); + var first = ExceptionFactory.PerMessage(shared, Read()); + var second = ExceptionFactory.PerMessage(shared, AccumulatingWrite()); + + Assert.NotSame(first, second); + Assert.Equal(shared.Message, first.Message); + Assert.Equal("1.2.3", first.Data["Redis-Version"]); + Assert.Equal("1.2.3", second.Data["Redis-Version"]); + + first.Data["mine"] = "only-mine"; + Assert.False(second.Data.Contains("mine")); + Assert.False(shared.Data.Contains("mine")); + } + + // The per-message status is recorded in the diagnostic data too, so a user reading the exception's Data + // sees this message's status rather than whatever the connection-level exception happened to say. + [Fact] + public void SentStatusIsRecordedInDiagnosticData() + { + var shared = SharedConnectionFault(); + shared.Data["request-sent-status"] = CommandStatus.Unknown; + + var message = Read(); + message.SetRequestSent(); + var per = ExceptionFactory.PerMessage(shared, message); + + Assert.Equal(CommandStatus.Sent, per.Data["request-sent-status"]); + } + + // Only the shared connection-failure shape needs splitting; anything else already describes a single + // operation, and an exception that already matches the message is passed straight through (no needless + // allocation on a teardown that may be failing thousands of messages). + [Fact] + public void UnrelatedOrAlreadyMatchingExceptions_ArePassedThrough() + { + var message = Read(); + message.SetRequestSent(); + + var serverFault = new RedisServerException(RedisErrorKind.Loading, message.Flags, "LOADING"); + Assert.Same(serverFault, ExceptionFactory.PerMessage(serverFault, message)); + + var alreadySpecific = new RedisConnectionException( + ConnectionFailureType.SocketClosed, + message.Flags, + "already describes this message", + innerException: null, + commandStatus: CommandStatus.Sent); + Assert.Same(alreadySpecific, ExceptionFactory.PerMessage(alreadySpecific, message)); + } +} diff --git a/tests/StackExchange.Redis.Tests/RetryTests/RetryControllerTests.cs b/tests/StackExchange.Redis.Tests/RetryTests/RetryControllerTests.cs new file mode 100644 index 000000000..313d40c30 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RetryTests/RetryControllerTests.cs @@ -0,0 +1,206 @@ +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using StackExchange.Redis.Availability; +using StackExchange.Redis.Interfaces; +using Xunit; + +namespace StackExchange.Redis.Tests.RetryTests; + +// Configuration validation (which lives on RetryPolicy.Builder, since a RetryPolicy is immutable and +// validated on construction) and the wait/failover timing state machine of RetryController; neither needs a +// server, or even an inner database - CanRetry and the delays never touch one. +public class RetryControllerTests +{ + // A failover threshold below 1 could never be reached by the attempt counter (which starts at 1), so + // it would *silently* disable failover; that is rejected up front. + [Fact] + public void Policy_RejectsUnreachableFailoverThreshold() + => Assert.Throws(() => new RetryPolicy.Builder { MaxAttemptsBeforeFailover = 0 }.Create()); + + // DatabaseFeatureFlags is internal, so theories take a bool and map here + private static DatabaseFeatureFlags Features(bool withFailover) + => withFailover ? DatabaseFeatureFlags.Failover : DatabaseFeatureFlags.None; + + // Negative durations are nonsense for a delay; each is validated separately. + [Fact] + public void Policy_RejectsNegativeDurations() + { + var negative = TimeSpan.FromMilliseconds(-1); + Assert.Throws(() => new RetryPolicy.Builder { RetryDelay = negative }.Create()); + Assert.Throws(() => new RetryPolicy.Builder { JitterMax = negative }.Create()); + Assert.Throws(() => new RetryPolicy.Builder { FailoverDelay = negative }.Create()); + } + + // The watch-contention budget counts *attempts*, so 1 means "try once, do not re-attempt"; zero or + // negative is meaningless rather than a way to say "never execute". + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Policy_RejectsNonPositiveWatchAttempts(int attempts) + => Assert.Throws(() => new RetryPolicy.Builder { MaxAttemptsOnWatchConflict = attempts }.Create()); + + // ...and 1 is accepted, since that is how re-attempting is switched off + [Fact] + public void Policy_AcceptsSingleWatchAttempt() + => Assert.Equal(1, new RetryPolicy.Builder { MaxAttemptsOnWatchConflict = 1 }.Create().MaxAttemptsOnWatchConflict); + + // The category cap must name exactly one of the CommandRetry* values: an empty value, or one that + // strays outside the category bits, is a usage error rather than something to interpret. + [Fact] + public void Policy_RejectsNonCategoryMaxCommandRetryCategory() + { + Assert.Throws(() => new RetryPolicy.Builder { MaxCommandRetryCategory = CommandFlags.None }.Create()); + Assert.Throws(() => new RetryPolicy.Builder { MaxCommandRetryCategory = CommandFlags.FireAndForget }.Create()); + Assert.Throws( + () => new RetryPolicy.Builder { MaxCommandRetryCategory = CommandFlags.CommandRetryReadOnly | CommandFlags.FireAndForget }.Create()); + + RetryPolicy valid = new RetryPolicy.Builder { MaxCommandRetryCategory = CommandFlags.CommandRetryAlways }; + Assert.Equal(CommandFlags.CommandRetryAlways, valid.MaxCommandRetryCategory); + } + + // RetryPolicy.None means *nothing* is re-attempted - including watch contention, which is bounded by + // an attempt count rather than by CanRetry (nothing was applied, so there is no fault to judge). + [Fact] + public void NonePolicy_DisablesWatchReattempts() + { + Assert.Equal(1, RetryPolicy.None.MaxAttemptsOnWatchConflict); + Assert.Equal(DefaultMaxAttemptsOnWatchConflict, RetryPolicy.Default.MaxAttemptsOnWatchConflict); + } + + private const int DefaultMaxAttemptsOnWatchConflict = 3; + + // Round-tripping a policy through a builder must preserve the watch budget along with everything else. + [Fact] + public void Policy_RoundTripsThroughBuilder() + { + RetryPolicy original = new RetryPolicy.Builder { MaxAttemptsOnWatchConflict = 7, MaxAttempts = 4 }; + var copy = new RetryPolicy.Builder(original).Create(); + + Assert.Equal(7, copy.MaxAttemptsOnWatchConflict); + Assert.Equal(4, copy.MaxAttempts); + } + + // Contention is not a fault, so there is no backoff - only jitter, to stop two callers colliding again + // in lock-step. With jitter disabled the re-attempt is immediate. + [Fact] + public async Task WatchConflictDelay_HasNoBackoff() + { + var controller = new RetryController( + new RetryPolicy.Builder + { + RetryDelay = TimeSpan.FromMilliseconds(LongMillis), + FailoverDelay = TimeSpan.FromMilliseconds(LongMillis), + JitterMax = TimeSpan.Zero, + }, + DatabaseFeatureFlags.Failover); + + var watch = Stopwatch.StartNew(); + await controller.WatchConflictDelayAsync(); + Assert.True(watch.ElapsedMilliseconds < ShortMillis, $"returned after {watch.ElapsedMilliseconds}ms"); + } + + // Capturing the "next failover" token costs something, so we only do it when a failover could + // actually be waited on: the database must offer failover, there must be more than one attempt, and + // the threshold must sit strictly below the attempt cap (at the cap it can never be reached). + [Theory] + [InlineData(3, 1, true, true)] + [InlineData(3, 1, false, false)] // no failover available + [InlineData(1, 1, true, false)] // single attempt: nothing to retry + [InlineData(3, 3, true, false)] // threshold == cap: unreachable + [InlineData(3, 4, true, false)] // threshold beyond cap: unreachable + public void TracksFailover_OnlyWhenReachable(int maxAttempts, int beforeFailover, bool withFailover, bool expected) + { + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = maxAttempts, MaxAttemptsBeforeFailover = beforeFailover }; + Assert.Equal(expected, new RetryController(policy, Features(withFailover)).TracksFailover); + } + + // MaxAttempts = 1 means "try once": the very first failure is already exhausted. + [Fact] + public void SingleAttempt_NeverRetries() + { + var controller = new RetryController(new RetryPolicy.Builder { MaxAttempts = 1 }, DatabaseFeatureFlags.Failover); + using var cts = new CancellationTokenSource(); + var failover = cts.Token; + var fault = new RedisServerException(RedisErrorKind.Loading, CommandFlags.CommandRetryReadOnly, "LOADING"); + + Assert.False(controller.CanRetry(1, fault, ref failover, out var delay)); + Assert.False(delay.CanBeCanceled); + } + + // --- FailoverOrDelayAsync ----------------------------------------------------------------------- + // Deliberately coarse thresholds: we are distinguishing "waited for the configured period" from + // "returned as soon as it could", not measuring the clock. + private const int LongMillis = 2000, ShortMillis = 1000; + + // No failover token: this is a routine pause between same-server attempts, so it waits RetryDelay. + [Fact] + public async Task Delay_WithoutFailoverToken_WaitsRetryDelay() + { + var controller = new RetryController( + new RetryPolicy.Builder { RetryDelay = TimeSpan.FromMilliseconds(LongMillis), JitterMax = TimeSpan.Zero }, + DatabaseFeatureFlags.None); + + var watch = Stopwatch.StartNew(); + await controller.FailoverOrDelayAsync(CancellationToken.None); + Assert.True(watch.ElapsedMilliseconds >= ShortMillis, $"returned after {watch.ElapsedMilliseconds}ms"); + } + + // A failover token that has *already* fired: there is nothing to wait for, so only jitter applies - + // and in particular RetryDelay is deliberately ignored on the failover path. + [Fact] + public async Task Delay_WithFiredFailoverToken_ReturnsImmediately() + { + var controller = new RetryController( + new RetryPolicy.Builder + { + RetryDelay = TimeSpan.FromMilliseconds(LongMillis), + FailoverDelay = TimeSpan.FromMilliseconds(LongMillis), + JitterMax = TimeSpan.Zero, + }, + DatabaseFeatureFlags.Failover); + + using var cts = new CancellationTokenSource(); + cts.Cancel(); // Cancel, not CancelAsync: this project also targets net481 + + var watch = Stopwatch.StartNew(); + await controller.FailoverOrDelayAsync(cts.Token); + Assert.True(watch.ElapsedMilliseconds < ShortMillis, $"returned after {watch.ElapsedMilliseconds}ms"); + } + + // A failover that arrives while we are waiting: we stop waiting as soon as it lands, rather than + // sitting out the whole FailoverDelay. + [Fact] + public async Task Delay_WhenFailoverArrives_StopsWaiting() + { + var controller = new RetryController( + new RetryPolicy.Builder { FailoverDelay = TimeSpan.FromMilliseconds(LongMillis * 4), JitterMax = TimeSpan.Zero }, + DatabaseFeatureFlags.Failover); + + using var cts = new CancellationTokenSource(); + var watch = Stopwatch.StartNew(); + var pending = controller.FailoverOrDelayAsync(cts.Token); + cts.Cancel(); // Cancel, not CancelAsync: this project also targets net481 + await pending; + + Assert.True(watch.ElapsedMilliseconds < ShortMillis, $"returned after {watch.ElapsedMilliseconds}ms"); + } + + // A failover that never arrives: we give it FailoverDelay and then proceed anyway (retrying on the + // original server is better than giving up). + [Fact] + public async Task Delay_WhenFailoverNeverArrives_ProceedsAfterFailoverDelay() + { + var controller = new RetryController( + new RetryPolicy.Builder { FailoverDelay = TimeSpan.FromMilliseconds(LongMillis), JitterMax = TimeSpan.Zero }, + DatabaseFeatureFlags.Failover); + + using var cts = new CancellationTokenSource(); + var watch = Stopwatch.StartNew(); + await controller.FailoverOrDelayAsync(cts.Token); + + Assert.True(watch.ElapsedMilliseconds >= ShortMillis, $"returned after {watch.ElapsedMilliseconds}ms"); + Assert.False(cts.IsCancellationRequested); // no failover ever happened + } +} diff --git a/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs b/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs index 414823af0..8f2b6b623 100644 --- a/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs +++ b/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs @@ -1,5 +1,6 @@ using System; using System.Net; +using System.Threading; using System.Threading.Tasks; using StackExchange.Redis.Availability; using StackExchange.Redis.Server; @@ -45,6 +46,99 @@ public async Task WithRetry_RidesOutTransientLoading() Assert.Equal(3, server.GetOpsReceived); // 2 x LOADING + 1 x success } + // Retries are bounded: once MaxAttempts is used up the original server fault is surfaced to the + // caller unchanged (not wrapped, not swallowed), and the server saw exactly MaxAttempts requests. + [Fact] + public async Task WithRetry_WhenAttemptsExhausted_ThrowsOriginalFault() + { + using var server = new LoadingServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:exhaust"; + Assert.True(await db.StringSetAsync(key, "hello")); + + server.LoadingOps = 100; // more than we will ever attempt + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + var ex = await Assert.ThrowsAsync(async () => await db.WithRetry(policy).StringGetAsync(key)); + + Assert.Equal(RedisErrorKind.Loading, ex.Kind); + Assert.Equal(3, server.GetOpsReceived); // tried exactly MaxAttempts times + } + + // A fault that will not fix itself is not worth repeating: WRONGTYPE is an application error, so it + // surfaces on the first attempt regardless of how many attempts the policy allows. + [Fact] + public async Task WithRetry_NonTransientFault_IsNotRetried() + { + using var server = new LoadingServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:wrongtype"; + Assert.True(await db.StringSetAsync(key, "hello")); + + server.LoadingOps = 100; + server.ErrorText = "WRONGTYPE Operation against a key holding the wrong kind of value"; + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + var ex = await Assert.ThrowsAsync(async () => await db.WithRetry(policy).StringGetAsync(key)); + + Assert.Equal(RedisErrorKind.WrongType, ex.Kind); + Assert.Equal(1, server.GetOpsReceived); // gave up immediately + } + + // An ad-hoc command whose *name* is recognised gets that command's category for free, so a plain + // Execute("get", ...) is retried like any other read - the caller does not have to say anything. + [Fact] + public async Task WithRetry_AdHocCommand_InheritsKnownCommandCategory() + { + using var server = new LoadingServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:adhoc:known"; + Assert.True(await db.StringSetAsync(key, "hello")); + + server.LoadingOps = 2; + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + var result = await db.WithRetry(policy).ExecuteAsync("get", [key]); + + Assert.Equal("hello", result.AsString()); + Assert.Equal(3, server.GetOpsReceived); // recognised as GET, i.e. read-only, so retried + } + + // A command the library does *not* recognise could do anything, so it is treated pessimistically and + // never retried; a caller who knows better can say so via the flags. + [Theory] + [InlineData(CommandFlags.None, 1)] // unrecognised: assume the worst, do not retry + [InlineData(CommandFlags.CommandRetryReadOnly, 3)] // caller asserts it is a pure read + public async Task WithRetry_UnrecognisedCommand_RespectsSuppliedCategory(CommandFlags flags, int expectedAttempts) + { + using var server = new LoadingServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + server.LoadingOps = 2; // the third attempt would succeed, if we are allowed a third + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + var retryDb = db.WithRetry(policy); + + if (expectedAttempts == 1) + { + await Assert.ThrowsAsync(async () => await retryDb.ExecuteAsync("notarealcommand", [], flags)); + } + else + { + var result = await retryDb.ExecuteAsync("notarealcommand", [], flags); + Assert.Equal("made-up-ok", result.AsString()); + } + + Assert.Equal(expectedAttempts, server.UnknownOpsReceived); + } + // Multi-group + WithRetry: two backends hold *different* values for the same key. The group is weighted // towards A, so a normal read returns A's value. When A drops into LOADING, the faults trip A's // (deliberately hair-trigger) circuit-breaker; the group reroutes to B, and the retry wrapper rides that @@ -125,6 +219,62 @@ public async Task WithRetry_FailsOverBetweenGroupsOnLoading() Assert.Same(members[1], conn.ActiveMember); // we really did move to B } + // The unhappy path of the failover threshold: the attempt count says "wait for a failover now", but no + // failover ever comes (the member stays nominally healthy - a couple of LOADING replies are not enough + // to trip the default breaker). We wait out FailoverDelay and then carry on retrying the original + // member rather than giving up, and the group never moves. + [Fact] + public async Task WithRetry_WhenFailoverNeverArrives_KeepsRetryingSameMember() + { + EndPoint alpha = new DnsEndPoint("alpha", 6379); + EndPoint bravo = new DnsEndPoint("bravo", 6379); + using var serverA = new LoadingServer(Output, endpoint: alpha); + using var serverB = new InProcessTestServer(Output, endpoint: bravo); + + RedisKey key = "retry:nofailover"; + await using (var seedA = await serverA.ConnectAsync()) + { + Assert.True(await seedA.GetDatabase().StringSetAsync(key, "from-A")); + } + + ConnectionGroupMember[] members = + [ + new(serverA.GetClientConfig(), "A") { Weight = 9 }, // highest weight -> active, and stays active + new(serverB.GetClientConfig(), "B") { Weight = 1 }, + ]; + + MultiGroupOptions options = new MultiGroupOptions.Builder + { + HealthCheckInterval = TimeSpan.FromMinutes(30), + HealthCheck = new HealthCheck.Builder + { + Probe = new ControllableProbe(), // never marked down + ProbeCount = 1, + ProbeTimeout = TimeSpan.FromSeconds(5), + }, + }; + + await using var conn = await ConnectionMultiplexer.ConnectGroupAsync(members, options); + Assert.Same(members[0], conn.ActiveMember); + + // failover is armed after the first attempt, but nothing will ever trigger it; keep the wait short + RetryPolicy policy = new RetryPolicy.Builder + { + MaxAttempts = 5, + MaxAttemptsBeforeFailover = 1, + FailoverDelay = TimeSpan.FromMilliseconds(200), + RetryDelay = TimeSpan.Zero, + JitterMax = TimeSpan.Zero, + }; + var db = conn.GetDatabase().WithRetry(policy); + + serverA.LoadingOps = 2; // two transient faults, then A answers normally + + Assert.Equal("from-A", await db.StringGetAsync(key)); + Assert.Equal(3, serverA.GetOpsReceived); // 2 x LOADING + 1 x success, all on A + Assert.Same(members[0], conn.ActiveMember); // never moved to B + } + // End-to-end for a retryable transaction: a server that fails the first EXEC with a transient LOADING // error (discarding that attempt's queued commands), then commits the second. A transaction created via // the retrying database should ride this out: the per-operation tasks handed out at build time resolve @@ -164,13 +314,14 @@ public async Task WithRetry_Transaction_RidesOutTransientExec() Assert.Equal(2, server.ExecOpsReceived); // 1 x LOADING + 1 x commit } - // The transaction's effective retry category is the most side-effecting of its operations. An INCR makes - // the whole transaction "accumulating", which the default policy (capped at write-last-wins) refuses to - // retry - so a transient EXEC failure surfaces and the per-op proxy faults rather than hanging. Raising - // the cap to allow accumulating writes lets the same transaction ride the failure out; and because the - // failed attempt is discarded server-side, the INCR applies exactly once (no double-count). + // A transaction's effective retry category is the most side-effecting of its operations, so an INCR + // makes the whole thing "accumulating" - beyond what the default policy (capped at write-last-wins) + // would normally repeat. But a LOADING reply to EXEC *proves* the server discarded the transaction + // wholesale, so replaying it cannot double-count: the category cap does not apply, and even the + // default policy rides it out. Without that distinction, most interesting transactions (i.e. the ones + // that mutate something) would never be retryable at all. [Fact] - public async Task WithRetry_Transaction_AccumulatingOp_RespectsCategoryGate() + public async Task WithRetry_Transaction_RejectedExec_RetriesRegardlessOfCategory() { using var server = new ExecFailServer(Output); await using var conn = await server.ConnectAsync(log: Writer); @@ -179,30 +330,52 @@ public async Task WithRetry_Transaction_AccumulatingOp_RespectsCategoryGate() var db = conn.GetDatabase(); RedisKey key = "retry:tran:incr"; - // default cap = write-last-wins; an INCR makes the aggregate accumulating -> NOT retried - RetryPolicy conservative = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + Assert.Equal(CommandFlags.CommandRetryWriteLastWins, policy.MaxCommandRetryCategory); // i.e. the default + server.FailExecOps = 1; - var tran1 = db.WithRetry(conservative).CreateTransaction(); - var incr1 = tran1.StringIncrementAsync(key); - await Assert.ThrowsAsync(async () => await tran1.ExecuteAsync()); - await Assert.ThrowsAsync(async () => await incr1); // proxy faulted, not left hanging - Assert.Equal(1, server.ExecOpsReceived); // gave up immediately, no retry - Assert.Equal(0, server.FailExecOps); - - // raise the cap to allow accumulating writes: the same transaction now rides out the transient failure - RetryPolicy permissive = new RetryPolicy.Builder + var tran = db.WithRetry(policy).CreateTransaction(); + var incr = tran.StringIncrementAsync(key); + + Assert.True(await tran.ExecuteAsync()); + Assert.Equal(1, await incr); // the discarded attempt applied nothing -> incremented exactly once + Assert.Equal(2, server.ExecOpsReceived); // 1 x LOADING + 1 x commit + } + + // The other half of that story: when the outcome is genuinely *ambiguous*, the category still bites. + // OOM is deliberately *not* treated as "known not applied": a Lua script can hit the memory limit + // part-way through, having already written something, and it reports the inner error verbatim - so from + // the client's side an OOM reply proves nothing about whether the command took effect. Here the server + // models exactly that: it applies the INCR and *then* reports OOM. Under the default cap the fault + // surfaces after one attempt; raising the cap opts into replaying it, and the value shows the + // triple-count that the cap exists to prevent. + [Theory] + [InlineData(false, 1)] // default cap: not repeated + [InlineData(true, 3)] // caller opted in: repeated, and every attempt landed + public async Task WithRetry_AmbiguousFault_IsStillGatedByCategory(bool allowAccumulating, int expectedValue) + { + using var server = new AppliedThenFailedServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:ambiguous:incr"; + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero, - MaxCommandRetryCategory = CommandFlags.CommandRetryWriteAccumulating, + MaxCommandRetryCategory = allowAccumulating + ? CommandFlags.CommandRetryWriteAccumulating + : RetryPolicy.Default.MaxCommandRetryCategory, }; - server.FailExecOps = 1; - var tran2 = db.WithRetry(permissive).CreateTransaction(); - var incr2 = tran2.StringIncrementAsync(key); - Assert.True(await tran2.ExecuteAsync()); - Assert.Equal(1, await incr2); // discarded attempt did NOT apply -> incremented exactly once - Assert.Equal(3, server.ExecOpsReceived); // 1 (first block) + 2 (LOADING + commit) + + var ex = await Assert.ThrowsAsync(async () => await db.WithRetry(policy).StringIncrementAsync(key)); + Assert.Equal(RedisErrorKind.OutOfMemory, ex.Kind); + + Assert.Equal(expectedValue, server.IncrOpsReceived); + server.FailIncr = false; + Assert.Equal(expectedValue, (long)await db.StringGetAsync(key)); // and each one really did apply } // A WATCH constraint is replayed on every attempt. Here the condition is satisfied, and the first EXEC @@ -372,14 +545,241 @@ public async Task WithRetry_Transaction_FailsOverBetweenGroups() Assert.Equal("committed-on-B", await checkB.GetDatabase().StringGetAsync(key)); } + // Replay has to be repeatable, not just possible once: two transient EXEC failures in a row, with a + // mixed bag of operations, and every proxy still resolves from the third (winning) attempt. + [Fact] + public async Task WithRetry_Transaction_ReplaysRepeatedly() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:repeat", counter = "retry:tran:repeat:count", list = "retry:tran:repeat:list"; + Assert.True(await db.StringSetAsync(key, "seed")); + + server.FailExecOps = 2; // fail twice; the third EXEC commits + + RetryPolicy policy = new RetryPolicy.Builder + { + MaxAttempts = 4, + RetryDelay = TimeSpan.Zero, + JitterMax = TimeSpan.Zero, + MaxCommandRetryCategory = CommandFlags.CommandRetryWriteAccumulating, // INCR/LPUSH are accumulating + }; + var tran = db.WithRetry(policy).CreateTransaction(); + var cond = tran.AddCondition(Condition.StringEqual(key, "seed")); + var set = tran.StringSetAsync(key, "committed"); + var incr = tran.StringIncrementAsync(counter); + var push = tran.ListLeftPushAsync(list, "item"); + var get = tran.StringGetAsync(key); + + Assert.True(await tran.ExecuteAsync()); + + Assert.True(cond.WasSatisfied); + Assert.True(await set); + Assert.Equal(1, await incr); // the two discarded attempts applied nothing + Assert.Equal(1, await push); + Assert.Equal("committed", await get); + Assert.Equal(3, server.ExecOpsReceived); // 2 x LOADING + 1 x commit + } + + // When a transaction runs out of attempts, the failure must reach *every* per-operation proxy as well + // as the ExecuteAsync caller; a proxy left unresolved would hang the caller forever. + [Fact] + public async Task WithRetry_Transaction_WhenAttemptsExhausted_FaultsEveryProxy() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:exhaust"; + + server.FailExecOps = 100; // never succeeds + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 2, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + var tran = db.WithRetry(policy).CreateTransaction(); + var set = tran.StringSetAsync(key, "never"); + var get = tran.StringGetAsync(key); + + var fault = await Assert.ThrowsAsync(async () => await tran.ExecuteAsync()); + Assert.Equal(RedisErrorKind.Loading, fault.Kind); + + // both proxies carry the same terminal fault rather than being left pending + Assert.Same(fault, await Assert.ThrowsAsync(async () => await set)); + Assert.Same(fault, await Assert.ThrowsAsync(async () => await get)); + Assert.Equal(2, server.ExecOpsReceived); // exactly MaxAttempts + } + + // WATCH drift under retry: the condition holds, so EXEC really is issued, but the server refuses it + // because a watched key moved. Nothing was applied and nothing faulted - we simply lost a race - so + // the transaction is re-attempted (re-reading the condition), and the second attempt commits. This is + // contention, not a fault, so the fault budget is untouched and the side-effect category is irrelevant. + [Fact] + public async Task WithRetry_Transaction_WatchDrift_IsReattempted() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + Assert.True(conn.IsConnected); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:drift"; + Assert.True(await db.StringSetAsync(key, "seed")); + + server.DriftKey = key; + server.DriftOps = 1; // the next EXEC observes a concurrent write to the watched key + + // MaxAttempts = 1: no *fault* retries at all, proving the watch budget is separate + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 1, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + var tran = db.WithRetry(policy).CreateTransaction(); + var cond = tran.AddCondition(Condition.StringEqual(key, "seed")); + var setTask = tran.StringSetAsync(key, "committed"); + + var execute = tran.ExecuteAsync(); + if (await Task.WhenAny(execute, Task.Delay(5000)) != execute) + { + Assert.Fail("ExecuteAsync never completed"); + } + + Assert.True(await execute); // rode out the lost race + Assert.True(cond.WasSatisfied); + Assert.True(await setTask); + Assert.False(tran.WasWatchConflict); // reports the *final* attempt: we got there in the end + Assert.Equal(2, server.ExecOpsReceived); // 1 x watch conflict + 1 x commit + Assert.Equal("committed", await db.StringGetAsync(key)); + } + + // Watch contention is bounded, and the bound is opt-out-able: with MaxAttemptsOnWatchConflict = 1 a + // conflict aborts exactly as it did before. ExecuteAsync reports false with every condition satisfied + // (which is what distinguishes drift from an elective abort), and the per-operation proxies are + // cancelled rather than left dangling - the case that used to hang the caller outright. + [Fact] + public async Task WithRetry_Transaction_WatchDrift_CanBeDisabled() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:drift:off"; + Assert.True(await db.StringSetAsync(key, "seed")); + + server.DriftKey = key; + server.DriftOps = 1; + + RetryPolicy policy = new RetryPolicy.Builder + { + MaxAttempts = 3, + MaxAttemptsOnWatchConflict = 1, // i.e. do not re-attempt on contention + RetryDelay = TimeSpan.Zero, + JitterMax = TimeSpan.Zero, + }; + var tran = db.WithRetry(policy).CreateTransaction(); + var cond = tran.AddCondition(Condition.StringEqual(key, "seed")); + var setTask = tran.StringSetAsync(key, "committed"); + + var execute = tran.ExecuteAsync(); + if (await Task.WhenAny(execute, Task.Delay(5000)) != execute) + { + Assert.Fail("ExecuteAsync never completed"); + } + + Assert.False(await execute); + Assert.True(cond.WasSatisfied); // the condition held; the server-side WATCH is what killed it + Assert.True(tran.WasWatchConflict); // ...and the caller can see exactly that + Assert.Equal(1, server.ExecOpsReceived); + await Assert.ThrowsAnyAsync(async () => await setTask); + Assert.Equal("seed", await db.StringGetAsync(key)); // nothing was applied + } + + // Contention that never clears must not loop forever: the server conflicts on every EXEC, so we give + // up after MaxAttemptsOnWatchConflict attempts and report the ordinary "did not commit" outcome. + [Fact] + public async Task WithRetry_Transaction_PersistentWatchDrift_GivesUp() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:drift:forever"; + Assert.True(await db.StringSetAsync(key, "seed")); + + server.DriftKey = key; + server.DriftOps = int.MaxValue; // every EXEC loses the race + + RetryPolicy policy = new RetryPolicy.Builder + { + MaxAttempts = 3, + MaxAttemptsOnWatchConflict = 4, + RetryDelay = TimeSpan.Zero, + JitterMax = TimeSpan.Zero, + }; + var tran = db.WithRetry(policy).CreateTransaction(); + var cond = tran.AddCondition(Condition.StringEqual(key, "seed")); + var setTask = tran.StringSetAsync(key, "committed"); + + Assert.False(await tran.ExecuteAsync()); + Assert.True(cond.WasSatisfied); + Assert.True(tran.WasWatchConflict); // still losing the race when we ran out of attempts + Assert.Equal(4, server.ExecOpsReceived); // bounded by MaxAttemptsOnWatchConflict + await Assert.ThrowsAnyAsync(async () => await setTask); + Assert.Equal("seed", await db.StringGetAsync(key)); + } + + // A transaction with no conditions has no WATCH, so it can never lose a watch race; the watch budget + // must not be spent on the ordinary "aborted" path. (Belt and braces: the aggregate outcome here is a + // clean commit, so this mostly guards against the budget logic firing on a false positive.) + [Fact] + public async Task WithRetry_Transaction_WithoutConditions_IgnoresWatchBudget() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:nocond"; + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 1, MaxAttemptsOnWatchConflict = 5, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + var tran = db.WithRetry(policy).CreateTransaction(); + var set = tran.StringSetAsync(key, "committed"); + + Assert.True(await tran.ExecuteAsync()); + Assert.True(await set); + Assert.Equal(1, server.ExecOpsReceived); + } + + // An in-proc server that *applies* each INCR and then reports OOM: the write happened, but the client + // has no way to know that. Counts the applications so a test can tell "the client gave up" from "the + // client repeated a write it could not account for". + private sealed class AppliedThenFailedServer(ITestOutputHelper? log) : InProcessTestServer(log) + { + private int _incrOpsReceived; + + public int IncrOpsReceived => Volatile.Read(ref _incrOpsReceived); + + public bool FailIncr { get; set; } = true; + + protected override TypedRedisValue Incr(RedisClient client, in RedisRequest request) + { + var applied = base.Incr(client, in request); + if (!FailIncr) return applied; + + Interlocked.Increment(ref _incrOpsReceived); + return TypedRedisValue.Error("OOM command not allowed when used memory > 'maxmemory'"); + } + } + // An in-proc server that fails the first FailExecOps EXEC operations with a transient LOADING error, - // discarding that attempt's queued commands so nothing is applied, then commits normally. + // discarding that attempt's queued commands so nothing is applied, then commits normally. It can + // also simulate WATCH drift (a concurrent write to DriftKey immediately before EXEC is processed), + // which is a clean server-side rejection rather than a fault. private sealed class ExecFailServer(ITestOutputHelper? log, EndPoint? endpoint = null) : InProcessTestServer(log, endpoint) { public int ExecOpsReceived { get; private set; } public int FailExecOps { get; set; } + public int DriftOps { get; set; } + + public RedisKey DriftKey { get; set; } + protected override TypedRedisValue Exec(RedisClient client, in RedisRequest request) { ExecOpsReceived++; @@ -391,6 +791,12 @@ protected override TypedRedisValue Exec(RedisClient client, in RedisRequest requ return TypedRedisValue.Error("LOADING Redis is loading the dataset in memory"); } + if (DriftOps > 0) + { + DriftOps--; + client.Touch(client.Database, DriftKey); // as if another connection wrote the watched key + } + return base.Exec(client, in request); } } @@ -398,7 +804,7 @@ protected override TypedRedisValue Exec(RedisClient client, in RedisRequest requ // An in-proc server that fails the first LoadingOps GET operations with a transient LOADING error // (decrementing the counter each time), then serves normally. Every GET bumps GetOpsReceived so the // test can confirm how many attempts actually reached the server. - private sealed class LoadingServer(ITestOutputHelper? log) : InProcessTestServer(log) + private sealed class LoadingServer(ITestOutputHelper? log, EndPoint? endpoint = null) : InProcessTestServer(log, endpoint) { // the server core processes operations under a lock (single-threaded, like Redis), so plain fields // are fine here @@ -406,6 +812,27 @@ private sealed class LoadingServer(ITestOutputHelper? log) : InProcessTestServer public int LoadingOps { get; set; } + // the error to reply with; transient by default, but overridable so the same harness can present + // a fault that is *not* worth retrying + public string ErrorText { get; set; } = "LOADING Redis is loading the dataset in memory"; + + public int UnknownOpsReceived { get; private set; } + + // a command the *client* cannot categorise; answered from the same LoadingOps budget so the retry + // decision is isolated to the category, not the error kind + public override TypedRedisValue OnUnknownCommand(RedisClient client, in RedisRequest request, ReadOnlySpan command) + { + UnknownOpsReceived++; + + if (LoadingOps > 0) + { + LoadingOps--; + return TypedRedisValue.Error(ErrorText); + } + + return TypedRedisValue.SimpleString("made-up-ok"); + } + protected override TypedRedisValue Get(RedisClient client, in RedisRequest request) { GetOpsReceived++; @@ -414,7 +841,7 @@ protected override TypedRedisValue Get(RedisClient client, in RedisRequest reque if (LoadingOps > 0) { LoadingOps--; - return TypedRedisValue.Error("LOADING Redis is loading the dataset in memory"); + return TypedRedisValue.Error(ErrorText); } return base.Get(client, in request); diff --git a/tests/StackExchange.Redis.Tests/RetryTests/RetryGuardTests.cs b/tests/StackExchange.Redis.Tests/RetryTests/RetryGuardTests.cs new file mode 100644 index 000000000..64ea62f7e --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RetryTests/RetryGuardTests.cs @@ -0,0 +1,182 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using StackExchange.Redis.Availability; +using StackExchange.Redis.KeyspaceIsolation; +using Xunit; + +namespace StackExchange.Redis.Tests.RetryTests; + +// The rejection rules and hand-written members of the retry wrappers: what WithRetry refuses to wrap, +// what a retrying transaction refuses to do, and the members that are deliberately *not* retried +// (status probes, routing lookups, streaming scans) and so are implemented by hand. +public class RetryGuardTests(ITestOutputHelper log) : TestBase(log) +{ + private static RetryPolicy Policy() => new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + + // Retrying inside a batch or a transaction makes no sense (the individual operations are not being + // dispatched yet), and retry cannot be nested. All three are refused at wrap time. + [Fact] + public async Task WithRetry_RefusesBatchTransactionAndNesting() + { + using var server = new InProcessTestServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + var db = conn.GetDatabase(); + + Assert.Throws(() => db.CreateBatch().WithRetry(Policy())); + Assert.Throws(() => db.CreateTransaction().WithRetry(Policy())); + Assert.Throws(() => db.WithRetry(Policy()).WithRetry(Policy())); + } + + // A database's asyncState is stamped onto the task produced by a single dispatch. A retrying database + // hands back its own task spanning however many attempts it takes, so it cannot carry that state - + // and silently dropping it would be worse than refusing. + [Fact] + public async Task WithRetry_RefusesAsyncState() + { + using var server = new InProcessTestServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + object state = new(); + var ex = Assert.Throws(() => conn.GetDatabase(0, state).WithRetry(Policy())); + Log(ex.Message); + + // ...including via a key-prefixed view of such a database, which inherits the inner state + Assert.Throws(() => conn.GetDatabase(0, state).WithKeyPrefix("p:").WithRetry(Policy())); + + // and the same applies to a transaction created *from* a retrying database + var retryDb = conn.GetDatabase().WithRetry(Policy()); + Assert.Throws(() => retryDb.CreateTransaction(state)); + + // no state: fine + Assert.NotNull(conn.GetDatabase().WithKeyPrefix("p:").WithRetry(Policy())); + Assert.NotNull(retryDb.CreateTransaction()); + } + + // Key-prefixing composes with retry: the prefix is applied when the operation is captured, so it + // survives being replayed. (Only one nesting order is expressible, since WithKeyPrefix needs an + // IDatabase and WithRetry produces an async-only database.) + [Fact] + public async Task WithRetry_ComposesWithKeyPrefix() + { + using var server = new InProcessTestServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase().WithKeyPrefix("pfx:").WithRetry(Policy()); + Assert.True(await db.StringSetAsync("key", "value")); + + // visible under the prefixed name from an unprefixed database + Assert.Equal("value", await conn.GetDatabase().StringGetAsync("pfx:key")); + } + + // The transaction lifecycle guards: execute once, and do not accept work afterwards. + [Fact] + public async Task RetryTransaction_RefusesReuse() + { + using var server = new InProcessTestServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var tran = conn.GetDatabase().WithRetry(Policy()).CreateTransaction(); + var set = tran.StringSetAsync("guard:key", "value"); + Assert.True(await tran.ExecuteAsync()); + Assert.True(await set); + + await Assert.ThrowsAsync(async () => await tran.ExecuteAsync()); + Assert.Throws(() => { _ = tran.StringSetAsync("guard:key", "again"); }); + Assert.Throws(() => tran.AddCondition(Condition.KeyExists("guard:key"))); + } + + // Nested transactions, and the cursored scans, cannot participate in a transaction at all. + [Fact] + public async Task RetryTransaction_RefusesNestingAndScans() + { + using var server = new InProcessTestServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var tran = conn.GetDatabase().WithRetry(Policy()).CreateTransaction(); + + Assert.Throws(() => tran.CreateTransaction()); + Assert.Throws(() => tran.HashScanAsync("k")); + Assert.Throws(() => tran.HashScanNoValuesAsync("k")); + Assert.Throws(() => tran.SetScanAsync("k")); + Assert.Throws(() => tran.SortedSetScanAsync("k")); + Assert.Throws(() => tran.VectorSetRangeEnumerateAsync("k", "a", "z")); + } + + // Scans *can* be used on a retrying database: they are cursored, so they cannot be captured and + // replayed as a unit, but rather than refusing them outright we forward straight through (giving up + // retry, keeping the scan working). Needs a real server: the managed one has no *SCAN support. + [Fact] + public async Task WithRetry_ForwardsScans() + { + await using var conn = Create(); + + var inner = conn.GetDatabase(); + RedisKey hash = Me() + ":hash", set = Me() + ":set", zset = Me() + ":zset"; + await inner.KeyDeleteAsync([hash, set, zset]); + await inner.HashSetAsync(hash, [new HashEntry("a", "1"), new HashEntry("b", "2")]); + await inner.SetAddAsync(set, ["x", "y"]); + await inner.SortedSetAddAsync(zset, [new SortedSetEntry("p", 1), new SortedSetEntry("q", 2)]); + + var db = inner.WithRetry(Policy()); + + Assert.Equal(2, await CountAsync(db.HashScanAsync(hash))); + Assert.Equal(2, await CountAsync(db.HashScanNoValuesAsync(hash))); + Assert.Equal(2, await CountAsync(db.SetScanAsync(set))); + Assert.Equal(2, await CountAsync(db.SortedSetScanAsync(zset))); + } + + // Commands with no return value (a bare Task, not Task) go through their own funnel in the retry + // database, and their own recorded-operation type inside a retrying transaction. Needs a real server: + // the managed one implements no void-shaped command, so the fault/replay variant of this cannot + // currently be driven in-process. + [Fact] + public async Task WithRetry_HandlesVoidOperations() + { + await using var conn = Create(); + + var inner = conn.GetDatabase(); + RedisKey key = Me(); + await inner.KeyDeleteAsync(key); + await inner.ListRightPushAsync(key, ["a", "b", "c", "d"]); + + var db = inner.WithRetry(Policy()); + await db.ListTrimAsync(key, 0, 1); // Task, not Task + Assert.Equal(2, await db.ListLengthAsync(key)); + + // and the same shape recorded into (and replayed by) a retrying transaction + var tran = db.CreateTransaction(); + var trim = tran.ListTrimAsync(key, 0, 0); + var length = tran.ListLengthAsync(key); + Assert.True(await tran.ExecuteAsync()); + + await trim; // the void proxy resolved rather than hanging + Assert.True(trim.IsCompletedSuccessfully); + Assert.Equal(1, await length); + } + + // The cheap status/routing members are pass-throughs rather than retried round-trips. + [Fact] + public async Task WithRetry_ForwardsProbes() + { + using var server = new InProcessTestServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var inner = conn.GetDatabase(); + var db = inner.WithRetry(Policy()); + RedisKey key = "probe:key"; + + Assert.Equal(inner.Database, db.Database); + Assert.Same(conn, db.Multiplexer); + Assert.True(db.IsConnected(key)); + Assert.NotNull(await db.IdentifyEndpointAsync(key)); + Log(db.ToString()!); // exercises the feature-flag description + } + + private static async Task CountAsync(IAsyncEnumerable source) + { + int count = 0; + await foreach (var _ in source) count++; + return count; + } +} diff --git a/tests/StackExchange.Redis.Tests/TransactionWatchDriftTests.cs b/tests/StackExchange.Redis.Tests/TransactionWatchDriftTests.cs new file mode 100644 index 000000000..7b4f1bc26 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/TransactionWatchDriftTests.cs @@ -0,0 +1,189 @@ +using System.Net; +using System.Threading.Tasks; +using RESPite.Messages; +using StackExchange.Redis.Server; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Covers the "WATCH drift" outcome of a conditional transaction: every condition was satisfied, so +/// MULTI/EXEC really was issued, but a watched key changed underneath us and the server +/// answered EXEC with a null array. This is distinct from an *elective* abort (a condition that +/// failed, where no EXEC is sent at all) and, unlike that case, it can only be produced by a +/// concurrent write - so it needs the in-process server to drive it deterministically. +/// +[RunPerProtocol] +public class TransactionWatchDriftTests(ITestOutputHelper log) : TestBase(log) +{ + // A null array is not an empty array: EXEC answering *-1 (RESP2) / _ (RESP3) means "watch failed", + // whereas *0 means "a transaction of zero commands committed". The managed server used to collapse + // the former into the latter, which made the whole drift path untestable (and, client-side, it + // surfaced as a protocol failure instead). + [Fact] + public void NullArrayIsDistinctFromEmptyArray() + { + var nullArray = TypedRedisValue.NullArray(RespPrefix.Array); + Assert.True(nullArray.IsNullArray); + Assert.True(nullArray.IsNullValueOrArray); + Assert.True(nullArray.Span.IsEmpty); + + var emptyArray = TypedRedisValue.EmptyArray(RespPrefix.Array); + Assert.False(emptyArray.IsNullArray); + Assert.False(emptyArray.IsNullValueOrArray); + Assert.True(emptyArray.Span.IsEmpty); + } + + // The headline case: the condition holds, EXEC is issued, and the server rejects it because the + // watched key moved. Execute reports false (nothing was applied) and - the part that regressed - + // every queued operation's task must reach a terminal state (cancelled), not hang forever. + [Fact] + public async Task WatchDrift_AbortsAndCancelsQueuedOperations() + { + using var server = new WatchDriftServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "drift:cancel"; + Assert.True(await db.StringSetAsync(key, "seed")); + + server.DriftKey = key; + server.DriftOps = 1; // the next EXEC observes a concurrent write to the watched key + + var tran = db.CreateTransaction(); + var cond = tran.AddCondition(Condition.StringEqual(key, "seed")); + var setTask = tran.StringSetAsync(key, "committed"); + var incrTask = tran.StringIncrementAsync("drift:counter"); + + Assert.False(await tran.ExecuteAsync()); // EXEC returned a null array + Assert.True(cond.WasSatisfied); // the *condition* held; the server-side WATCH is what killed it + Assert.True(tran.WasWatchConflict); // ...and this is how a caller tells the two apart + Assert.Equal(1, server.ExecOpsReceived); + + // both per-operation tasks must complete (as cancelled); before the fix they sat forever in + // WaitingForActivation, so assert with a timeout rather than awaiting them directly + await AssertCancelledAsync(setTask); + await AssertCancelledAsync(incrTask); + + Assert.Equal("seed", await db.StringGetAsync(key)); // nothing was applied + Assert.False(await db.KeyExistsAsync("drift:counter")); + } + + // Same shape, but confirming the *elective* abort still behaves: the condition fails, no EXEC is + // ever issued, and the queued operations are cancelled. This is the path that already worked; it is + // here so the two outcomes are pinned side by side (they are indistinguishable from Execute's bool + // alone - WasWatchConflict, or inspecting the ConditionResults, is what separates them). + [Fact] + public async Task FailedCondition_AbortsElectively_WithoutExec() + { + using var server = new WatchDriftServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "drift:elective"; + Assert.True(await db.StringSetAsync(key, "seed")); + + var tran = db.CreateTransaction(); + Assert.False(tran.WasWatchConflict); // false before execution, too + + var cond = tran.AddCondition(Condition.StringEqual(key, "different")); + var setTask = tran.StringSetAsync(key, "committed"); + + Assert.False(await tran.ExecuteAsync()); + Assert.False(cond.WasSatisfied); // this is what distinguishes an elective abort from drift + Assert.False(tran.WasWatchConflict); // we chose not to issue an EXEC; nobody raced us + Assert.Equal(0, server.ExecOpsReceived); // never even asked + + await AssertCancelledAsync(setTask); + Assert.Equal("seed", await db.StringGetAsync(key)); + } + + // A transaction with conditions but no operations: drift still aborts it, and there are no + // per-operation tasks to cancel. Guards the zero-length inner-operations edge in the processor. + [Fact] + public async Task WatchDrift_ConditionOnlyTransaction_Aborts() + { + using var server = new WatchDriftServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "drift:condonly"; + Assert.True(await db.StringSetAsync(key, "seed")); + + server.DriftKey = key; + server.DriftOps = 1; + + var tran = db.CreateTransaction(); + var cond = tran.AddCondition(Condition.StringEqual(key, "seed")); + + Assert.False(await tran.ExecuteAsync()); + Assert.True(cond.WasSatisfied); + Assert.True(tran.WasWatchConflict); + Assert.Equal(1, server.ExecOpsReceived); + } + + // A transaction that commits cleanly must not report a conflict. + [Fact] + public async Task SatisfiedCondition_Commits_WithoutConflict() + { + using var server = new WatchDriftServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "drift:clean"; + Assert.True(await db.StringSetAsync(key, "seed")); + + var tran = db.CreateTransaction(); + var cond = tran.AddCondition(Condition.StringEqual(key, "seed")); + var setTask = tran.StringSetAsync(key, "committed"); + + Assert.True(await tran.ExecuteAsync()); + Assert.True(cond.WasSatisfied); + Assert.False(tran.WasWatchConflict); + Assert.True(await setTask); + } + + private async Task AssertCancelledAsync(Task task) + { + var completed = await Task.WhenAny(task, Task.Delay(5000)); + if (completed != task) + { + Log($"task did not complete; status: {task.Status}"); + Assert.Fail($"queued operation never completed (status: {task.Status})"); + } + + await Assert.ThrowsAnyAsync(async () => await task); + } + + // An in-process server that, for the next DriftOps EXEC operations, simulates a concurrent write to + // DriftKey immediately before the EXEC is processed. Touch is exactly what a real write from another + // connection would do, so the transaction is doomed by the server's own WATCH bookkeeping and EXEC + // replies with a null array - no special-casing of the reply itself. + // + // Driving this from a genuinely separate connection is not practical: SE.Redis does not issue the WATCH + // when AddCondition is called, it issues WATCH, the condition reads, MULTI, the queued commands and + // EXEC as one dispatch. So the window an interloper has to squeeze into is the gap between the + // condition reads and the EXEC landing, within a single flush - which is the point of the feature, but + // makes it useless as a test lever. Injecting the Touch server-side reproduces the same state exactly. + private sealed class WatchDriftServer(ITestOutputHelper? log, EndPoint? endpoint = null) : InProcessTestServer(log, endpoint) + { + public int ExecOpsReceived { get; private set; } + + public int DriftOps { get; set; } + + public RedisKey DriftKey { get; set; } + + protected override TypedRedisValue Exec(RedisClient client, in RedisRequest request) + { + ExecOpsReceived++; + + if (DriftOps > 0) + { + DriftOps--; + client.Touch(client.Database, DriftKey); + } + + return base.Exec(client, in request); + } + } +} diff --git a/toys/StackExchange.Redis.Server/TypedRedisValue.cs b/toys/StackExchange.Redis.Server/TypedRedisValue.cs index d3264a1ec..da5d33a42 100644 --- a/toys/StackExchange.Redis.Server/TypedRedisValue.cs +++ b/toys/StackExchange.Redis.Server/TypedRedisValue.cs @@ -194,14 +194,17 @@ private TypedRedisValue(TypedRedisValue[] oversizedItems, int count, RespPrefix if (oversizedItems == null) { if (count != 0) throw new ArgumentOutOfRangeException(nameof(count)); - oversizedItems = []; - } - else - { - if (count < 0 || count > oversizedItems.Length) throw new ArgumentOutOfRangeException(nameof(count)); - if (count == 0) oversizedItems = []; + + // a *null* array is not the same as an empty array; keep the value null so that + // IsNullArray reports true and we emit *-1 / _ rather than *0 + _value = RedisValue.Null; + Type = type; + return; } + if (count < 0 || count > oversizedItems.Length) throw new ArgumentOutOfRangeException(nameof(count)); + if (count == 0) oversizedItems = []; + _value = RedisValue.CreateForeign(oversizedItems, 0, count); Type = type; }