diff --git a/docs/rules/SER352.md b/docs/rules/SER352.md deleted file mode 100644 index b4c0dd7be..000000000 --- a/docs/rules/SER352.md +++ /dev/null @@ -1,35 +0,0 @@ -# SER352: generated database members are not implemented - -A type marked `[AutoDatabase(WarnIfIncomplete = true)]` still has members that the generator supplied, and -those members throw rather than doing anything. The count is how many. - -This is a **transition tripwire**, not a code-quality rule. `[AutoDatabase]` fills in every member of -`IDatabase`/`IDatabaseAsync` that the type does not implement itself, funnelling them through the type's own -`Execute`/`ExecuteAsync`. For most users of the attribute those generated members genuinely work - they -forward to an inner database - and this rule is off for them. It is opt-in precisely because "generated" and -"unimplemented" are only the same thing for a type whose funnel throws, which is the shape of a type being -migrated to a new implementation one command at a time. - -## Fixing it - -Implement the member on the type. That is the whole fix, and it is self-maintaining: the generator skips -whatever the class declares itself, so the member stops being generated and the count goes down. When the -count reaches zero the warning disappears on its own. - -There is nothing to add to a list, and no list to forget to update - which is the reason the generated half -exists at all. - -## Why Release only - -The inner development loop is exactly when the type is *expected* to be incomplete, and a warning that fires -on every build during the work it is describing is a warning people learn to skip past. It is reported when -the `DEBUG` preprocessor symbol is absent, which is how the SDK's own `Release` configuration is defined. - -## Suppressing - -Do not suppress it to make it quiet - it is counting something real, and something that will throw in front -of a user. - -Do consider `SER352` while the migration is in progress: in a -build with `TreatWarningsAsErrors` this rule is otherwise fatal, which blocks the very work it is tracking. -Removing that line once the migration is close is what makes shipping an incomplete type impossible. diff --git a/docs/rules/index.md b/docs/rules/index.md index 1bb6322ee..bb44ab607 100644 --- a/docs/rules/index.md +++ b/docs/rules/index.md @@ -51,7 +51,6 @@ Unlike everything under [Usage](#usage), these describe code that does not do wh - [SER350](SER350) - language version too low for generated code - [SER351](SER351) - a `[Resp]` declaration the fragment generator cannot implement -- [SER352](SER352) - generated database members that are not implemented and will throw ## When these rules stay quiet diff --git a/eng/StackExchange.Redis.Build/AnalyzerReleases.Unshipped.md b/eng/StackExchange.Redis.Build/AnalyzerReleases.Unshipped.md index e0733f289..1e5213fe0 100644 --- a/eng/StackExchange.Redis.Build/AnalyzerReleases.Unshipped.md +++ b/eng/StackExchange.Redis.Build/AnalyzerReleases.Unshipped.md @@ -18,4 +18,3 @@ SER308 | Usage | Warning | QueuedResultAnalyzer: calling the library's own SER309 | Usage | Warning | RespInterpolationAnalyzer: literal text in a RESP interpolated command is parsed and encoded on every call, where a fragment or resolved command is prepared once SER310 | Usage | Info | CancellationAnalyzer: a redis call that cannot honour the CancellationToken parameter its enclosing method accepts SER351 | Build | Warning | RespFragmentGenerator: a [Resp] declaration that cannot be implemented, which would otherwise be skipped silently -SER352 | Build | Warning | AutoDatabaseGenerator: an [AutoDatabase(WarnIfIncomplete = true)] type still has generated members that only throw; Release builds only diff --git a/eng/StackExchange.Redis.Build/AutoDatabaseGenerator.cs b/eng/StackExchange.Redis.Build/AutoDatabaseGenerator.cs deleted file mode 100644 index 5ba2e7a28..000000000 --- a/eng/StackExchange.Redis.Build/AutoDatabaseGenerator.cs +++ /dev/null @@ -1,722 +0,0 @@ -using System; -using System.Collections.Immutable; -using System.Reflection.Metadata.Ecma335; -using System.Text; -using Microsoft.CodeAnalysis; -using Microsoft.CodeAnalysis.CSharp.Syntax; - -// readable names for the info we gather; kept as tuples (+ BasicArray) so they have -// value equality and play nicely with incremental generator caching. -using ParamInfo = (string Name, string Type, Microsoft.CodeAnalysis.RefKind RefKind, bool IsParams, bool IsOptional, bool HasDefault, string? Default); -using MethodInfo = (string Name, string ReturnType, StackExchange.Redis.Build.BasicArray<(string Name, string Type, Microsoft.CodeAnalysis.RefKind RefKind, bool IsParams, bool IsOptional, bool HasDefault, string? Default)> Parameters, StackExchange.Redis.Build.BasicArray TypeArgs); -using InterfaceInfo = (string Name, string Namespace, StackExchange.Redis.Build.AutoDatabaseGenerator.KnownInterfaces KnownType, StackExchange.Redis.Build.BasicArray<(string Name, string ReturnType, StackExchange.Redis.Build.BasicArray<(string Name, string Type, Microsoft.CodeAnalysis.RefKind RefKind, bool IsParams, bool IsOptional, bool HasDefault, string? Default)> Parameters, StackExchange.Redis.Build.BasicArray TypeArgs)> Methods); -using ClassInfo = (string Name, string Namespace, StackExchange.Redis.Build.AutoDatabaseGenerator.KnownInterfaces Interfaces, bool IsMutator, bool Replays, StackExchange.Redis.Build.BasicArray Declared, bool WarnIfIncomplete); - -namespace StackExchange.Redis.Build; - -[Generator(LanguageNames.CSharp)] -public class AutoDatabaseGenerator : IIncrementalGenerator -{ - public void Initialize(IncrementalGeneratorInitializationContext ctx) - { - var interfaces = ctx.SyntaxProvider - .CreateSyntaxProvider( - static (node, _) => node is InterfaceDeclarationSyntax decl && FastIndexFilter(decl), ExtractInterfaceMethods) - .Where(pair => pair.Name is { Length: > 0 }) - .Collect(); - - var classes = ctx.SyntaxProvider - .CreateSyntaxProvider( - static (node, _) => node is ClassDeclarationSyntax decl && FastClassFilter(decl), ExtractClasses) - .Where(pair => pair.Name is { Length: > 0 }) - .Collect(); - - // Release is "no DEBUG symbol", which is what the SDK's own configuration does; a generator has no - // other reliable view of $(Configuration) without the project opting the property in. - var isRelease = ctx.ParseOptionsProvider.Select( - static (options, _) => !options.PreprocessorSymbolNames.Contains("DEBUG")); - - ctx.RegisterSourceOutput( - interfaces.Combine(classes).Combine(isRelease), - static (ctx, content) => Generate(ctx, content.Left.Left, content.Left.Right, content.Right)); - } - - /// - /// The only assembly this generator has anything to say about. - /// - /// - /// This is repo-internal machinery, but it now ships as an analyzer inside the StackExchange.Redis package - /// (for AsciiHashGenerator's benefit), so it is loaded by every consumer. It can never generate - /// anything useful for them - the semantic checks below reject anything that isn't our own - /// StackExchange.Redis declaration - so short-circuit on the assembly name first, and skip even the - /// semantic-model query for the consumers who happen to declare a type called IDatabase. - /// - private const string OwningAssembly = "StackExchange.Redis"; - - static KnownInterfaces Identify(string type) => type switch - { - "IDatabase" => KnownInterfaces.IDatabase, - "IDatabaseAsync" => KnownInterfaces.IDatabaseAsync, - "IRedis" => KnownInterfaces.IRedis, - "IRedisAsync" => KnownInterfaces.IRedisAsync, - _ => KnownInterfaces.None, - }; - - private static bool FastIndexFilter(InterfaceDeclarationSyntax decl) // limit to IDatabase, IDatabaseAsync - => Identify(decl.Identifier.ValueText) is not 0; - - private static bool FastClassFilter(ClassDeclarationSyntax decl) // limit to IDatabase, IDatabaseAsync - => decl.AttributeLists.Any(x => x.Attributes.Any(x => x.Name.ToString() is "AutoDatabase" or "AutoDatabaseAttribute")); - - // does this auto-database rewrite keys/channels? if not, the captured-args structs need no mapping members - private static bool IsMutatorInterface(INamedTypeSymbol symbol) => - symbol is - { - TypeKind: TypeKind.Interface, - Name: "IRedisArgsMutator", - ContainingType: null, - ContainingNamespace: - { - Name: "Redis", - ContainingNamespace: - { - Name: "StackExchange", - ContainingNamespace.IsGlobalNamespace: true - } - } - }; - - private static bool IsOurInterface(INamedTypeSymbol symbol) => - symbol is - { - TypeKind: TypeKind.Interface, - Name: "IDatabase" or "IDatabaseAsync" or "IRedis" or "IRedisAsync", - ContainingType: null, - ContainingNamespace: - { - Name: "Redis", - ContainingNamespace: - { - Name: "StackExchange", - ContainingNamespace.IsGlobalNamespace: true - } - } - }; - - private static InterfaceInfo ExtractInterfaceMethods(GeneratorSyntaxContext context, CancellationToken cancel) - { - if (context.SemanticModel.Compilation.AssemblyName is not OwningAssembly) return default; - - // note: we deliberately do NOT interpret anything here - just capture the raw shape of every - // method (name, return type, and per-parameter name/type/modifiers/optionality/default) so that - // later passes have everything they might need. - if (context.SemanticModel.GetDeclaredSymbol(context.Node, cancel) is not INamedTypeSymbol iface - || !IsOurInterface(iface)) - { - return default; - } - - var knownType = Identify(iface.Name); - if (knownType is KnownInterfaces.None) return default; - - var methods = new List(); - foreach (var member in iface.GetMembers()) - { - cancel.ThrowIfCancellationRequested(); - if (member is not IMethodSymbol { MethodKind: MethodKind.Ordinary } method) continue; - - var parameters = BasicArray.From(method.Parameters, static p => ( - Name: p.Name, - Type: p.Type.ToDisplayString(SymbolDisplayFormat.CSharpErrorMessageFormat), - RefKind: p.RefKind, - IsParams: p.IsParams, - IsOptional: p.IsOptional, - HasDefault: p.HasExplicitDefaultValue, - Default: p.HasExplicitDefaultValue ? FormatDefault(p.ExplicitDefaultValue) : null)); - - BasicArray typeArgs = default; - if (method.IsGenericMethod) - { - typeArgs = BasicArray.From(method.TypeParameters, p => p.Name); - } - methods.Add(( - Name: method.Name, - ReturnType: method.ReturnType.ToDisplayString(SymbolDisplayFormat.CSharpErrorMessageFormat), - Parameters: parameters, - TypeArgs: typeArgs)); - } - - var ns = iface.ContainingNamespace.ToDisplayString(SymbolDisplayFormat.CSharpErrorMessageFormat); - return (iface.Name, ns, knownType, BasicArray.From(methods)); - } - - private ClassInfo ExtractClasses(GeneratorSyntaxContext context, CancellationToken cancel) - { - if (context.SemanticModel.Compilation.AssemblyName is not OwningAssembly) return default; - - // note: we deliberately do NOT interpret anything here - just capture the raw shape of every - // method (name, return type, and per-parameter name/type/modifiers/optionality/default) so that - // later passes have everything they might need. - if (context.SemanticModel.GetDeclaredSymbol(context.Node, cancel) is not INamedTypeSymbol cls - || !HasAutoDatabaseAttrib(cls)) - { - return default; - } - - static bool HasAutoDatabaseAttrib(INamedTypeSymbol symbol) - { - var attribs = symbol.GetAttributes(); - foreach (var attrib in attribs) - { - if (attrib.AttributeClass is - { - Name: "AutoDatabaseAttribute", - ContainingType: null, - ContainingNamespace: - { - Name: "Redis", - ContainingNamespace: - { - Name: "StackExchange", - ContainingNamespace.IsGlobalNamespace: true - } - } - }) - { - return true; - } - } - - return false; - } - - // AllInterfaces, not Interfaces: IBatch and ITransaction ARE IDatabaseAsync (IBatch : IDatabaseAsync, - // ITransaction : IBatch), so a class declaring one of those implements every async command member - // and wants exactly the same generated set. Matching only the directly-declared name generated - // nothing for them, silently - the class compiled, and the members it had not written were simply - // absent rather than throwing. The union below is idempotent, so a class that declares IDatabase - // directly is unaffected by also seeing it here transitively. - KnownInterfaces known = 0; - foreach (var iFace in cls.AllInterfaces) - { - if (IsOurInterface(iFace)) - { - switch (iFace.Name) - { - case "IDatabase": - known |= KnownInterfaces.IDatabase | KnownInterfaces.IDatabaseAsync | - KnownInterfaces.IRedis | KnownInterfaces.IRedisAsync; - break; - case "IDatabaseAsync": - known |= KnownInterfaces.IDatabaseAsync | KnownInterfaces.IRedisAsync; - break; - case "IRedis": - known |= KnownInterfaces.IRedis | KnownInterfaces.IRedisAsync; - break; - case "IRedisAsync": - known |= KnownInterfaces.IRedisAsync; - break; - } - } - } - - if (known is KnownInterfaces.None) return default; // nothing to do! - - // AllInterfaces (not Interfaces) so an inherited/explicit implementation still counts - bool isMutator = false; - foreach (var iFace in cls.AllInterfaces) - { - if (IsMutatorInterface(iFace)) - { - isMutator = true; - break; - } - } - - // [AutoDatabase(Replays = true)] - the owning database can invoke a captured operation more than - // once, so captured Memory arguments have to be copies rather than the caller's own buffer - bool replays = false, warnIfIncomplete = false; - foreach (var attrib in cls.GetAttributes()) - { - if (attrib.AttributeClass?.Name is not ("AutoDatabaseAttribute" or "AutoDatabase")) continue; - foreach (var named in attrib.NamedArguments) - { - if (named.Key == "Replays" && named.Value.Value is bool b) replays = b; - if (named.Key == "WarnIfIncomplete" && named.Value.Value is bool w) warnIfIncomplete = w; - } - } - - // What the class already implements by hand, so generation can stand back. The generator emits - // EXPLICIT interface implementations, which means a hand-written member does NOT collide - both - // compile, and interface dispatch silently prefers the generated one. So this is a correctness - // check, not a convenience: without it, a partial that implements StringGet for real would still - // see IDatabase.StringGet route to the funnel. - // - // Other partials are ordinary source and therefore visible here; this generator's own output is - // not part of the compilation it sees, so there is nothing to exclude. - var declared = new List(); - foreach (var member in cls.GetMembers()) - { - cancel.ThrowIfCancellationRequested(); - if (member is not IMethodSymbol { MethodKind: MethodKind.Ordinary or MethodKind.ExplicitInterfaceImplementation } method) continue; - declared.Add(SignatureKey(method)); - } - - var ns = cls.ContainingNamespace.ToDisplayString(SymbolDisplayFormat.CSharpErrorMessageFormat); - return (cls.Name, ns, known, isMutator, replays, BasicArray.From(declared), warnIfIncomplete); - } - - /// - /// Name plus parameter types, formatted exactly as the interface side formats them so the two can be - /// compared as strings. - /// - /// - /// An explicit implementation's Name is the fully-qualified Namespace.IFace.Member, so - /// the implemented member's own name is used instead - a hand-written explicit implementation and a - /// hand-written public one mean the same thing here: "do not generate this". - /// - private static string SignatureKey(IMethodSymbol method) - { - var name = method.ExplicitInterfaceImplementations.Length > 0 - ? method.ExplicitInterfaceImplementations[0].Name - : method.Name; - - var sb = new StringBuilder(name).Append('('); - bool first = true; - foreach (var p in method.Parameters) - { - if (!first) sb.Append(','); - first = false; - sb.Append(p.Type.ToDisplayString(SymbolDisplayFormat.CSharpErrorMessageFormat)); - } - - return sb.Append(')').ToString(); - } - - /// The same key, from the interface-side shape. - private static string SignatureKey(MethodInfo method) - { - var sb = new StringBuilder(method.Name).Append('('); - bool first = true; - foreach (var p in method.Parameters.Span) - { - if (!first) sb.Append(','); - first = false; - sb.Append(p.Type); - } - - return sb.Append(')').ToString(); - } - - [Flags] - internal enum KnownInterfaces - { - None = 0, - IDatabase = 1, - IDatabaseAsync = 2, - IRedis = 4, - IRedisAsync = 8, - } - - // methods that don't fit the capture-and-replay shape are left for the caller to implement manually: - // - generic methods (open type args can't be captured into a concrete state struct) - // - the Wait family (synchronization over caller-supplied Tasks, not server calls) - // - IsConnected: a synchronous status probe that IDatabaseAsync carries; it would route through the - // sync Execute funnel, which an async-only database (e.g. RetryTransaction) doesn't provide - and a - // connection check shouldn't be retried in any case - // - IdentifyEndpoint / IdentifyEndpointAsync: a routing lookup ("which endpoint serves this key"), - // not a replayable server command; wrappers want their own behaviour here (e.g. a connection-group - // resolving against the currently-active member and returning null when there is none) - // - streaming returns (IEnumerable / IAsyncEnumerable) whose execution is deferred - // Memory/ReadOnlyMemory arguments still belong to the caller once the call returns; a database - // that replays has to capture a copy. Matched on the open generic prefix so a nullable or an array *of* - // them - neither of which is the caller's own buffer in the same way - does not match. - private static bool IsMemoryArg(string type) - => type.StartsWith("System.ReadOnlyMemory<", StringComparison.Ordinal) - || type.StartsWith("System.Memory<", StringComparison.Ordinal); - - // the element type, for the ArrayPool we rent the copy from - private static string MemoryElement(string type) - => type.Substring(type.IndexOf('<') + 1, type.Length - type.IndexOf('<') - 2); - - private static bool SkipMethod(MethodInfo method) - => !method.TypeArgs.IsEmpty - || method.Name.Contains("Wait") - || method.Name == "IsConnected" - || method.Name.Contains("IdentifyEndpoint") - // transaction/batch factories are not server round-trips and cannot be captured-and-replayed; - // the (few) databases that offer them implement them by hand - || method.Name == "CreateTransaction" - || method.Name == "CreateBatch" - || method.ReturnType.StartsWith("System.Collections.Generic.IEnumerable<", StringComparison.Ordinal) - || method.ReturnType.StartsWith("System.Collections.Generic.IAsyncEnumerable<", StringComparison.Ordinal); - - private const string TaskType = "System.Threading.Tasks.Task"; - - // Task / Task returns route through ExecuteAsync (its own retry policy); everything else - // uses Execute. - private static bool IsAsync(string returnType) - => returnType.StartsWith(TaskType, StringComparison.Ordinal); - - // the retry machinery unwraps the Task and only ever sees the inner result, so key/channel - // (un)mapping must be decided on T, not Task; this strips the Task<...> wrapper from an async - // return type. A bare (non-generic) Task has no result and is returned unchanged (as is any - // non-async return type), which is harmless since neither matches NeedsMap. - private static string StripTask(string returnType) - => IsAsync(returnType) && returnType.StartsWith(TaskType + "<", StringComparison.Ordinal) - ? returnType.Substring(TaskType.Length + 1, returnType.Length - TaskType.Length - 2) - : returnType; - - private static string FormatDefault(object? value) => value switch - { - null => "null", - string s => $"\"{s}\"", - bool b => b ? "true" : "false", - _ => value.ToString() ?? "null", - }; - - private static void Generate(SourceProductionContext ctx, ImmutableArray interfaces, ImmutableArray classes, bool isRelease) - { - if (interfaces.IsDefaultOrEmpty | classes.IsDefaultOrEmpty) return; // nothing to do - - // each interface is declared across several partial files, so we get one (identical) entry per - // partial declaration; structural equality lets us collapse those down to one per interface. - var iKeyed = new Dictionary(); - foreach (var t in interfaces) - { - if (t.KnownType is KnownInterfaces.None) continue; - if (!iKeyed.ContainsKey(t.KnownType)) iKeyed.Add(t.KnownType, t); - } - - var sb = new StringBuilder(); - var writer = new CodeWriter(sb); - writer.NewLine().Append("// "); - writer.NewLine().Append("// AutoDatabaseGenerator - explicit interface implementations that funnel every"); - writer.NewLine().Append("// call through Execute(state, projection), capturing the arguments in a generated"); - writer.NewLine().Append("// state struct (one per unique parameter-type signature) to avoid per-call closures."); - writer.NewLine().Append("#nullable enable"); // this needs to be explicit for code-gen - writer.NewLine(); - foreach (var cls in classes.Distinct()) - { - if (!string.IsNullOrWhiteSpace(cls.Namespace)) - { - writer.NewLine().Append("namespace ").Append(cls.Namespace).NewLine().Append("{").Indent(); - } - - // what this class implements by hand, and therefore must NOT have generated for it - var declared = new HashSet(StringComparer.Ordinal); - foreach (var signature in cls.Declared) declared.Add(signature); - - // how many members this class did NOT implement, for the SER352 tripwire below - int generated = 0; - - // unique parameter-type signatures encountered while emitting this class's methods; - // keyed on the '|'-joined parameter types so distinct methods with the same shape share - // one state struct. tupleDefs[i] holds a representative parameter list for _tuple{i}. - var tupleIndex = new Dictionary(); - var tupleDefs = new List>(); - - // every unique key/channel-bearing result type across all shapes; a single shared - // singleton (emitted after the tuples) implements IRedisArgsResult for each, so - // tuples expose unmap dispatch via UnMapper without ever being cast to an interface - var allReturns = new HashSet(); - - bool isFirst = true; - writer.NewLine().Append("partial class ").Append(cls.Name); - AppendInterfaceDeclaration(KnownInterfaces.IDatabase); - AppendInterfaceDeclaration(KnownInterfaces.IDatabaseAsync); - AppendInterfaceDeclaration(KnownInterfaces.IRedis); - AppendInterfaceDeclaration(KnownInterfaces.IRedisAsync); - writer.NewLine().Append("{").Indent(); - AppendInterfaceMethods(KnownInterfaces.IDatabase); - AppendInterfaceMethods(KnownInterfaces.IDatabaseAsync); - AppendInterfaceMethods(KnownInterfaces.IRedis); - AppendInterfaceMethods(KnownInterfaces.IRedisAsync); - AppendTupleTypes(); - - // The tripwire: a type opted into WarnIfIncomplete is mid-transition, and every member still - // generated for it is one the class has not taken over. Release only, so the inner loop is - // quiet and the noise arrives exactly when someone is preparing something that ships. - if (isRelease && cls.WarnIfIncomplete && generated > 0) - { - var fqn = string.IsNullOrWhiteSpace(cls.Namespace) ? cls.Name : cls.Namespace + "." + cls.Name; - ctx.ReportDiagnostic(Diagnostic.Create(Diagnostics.AutoDatabaseIncomplete, Location.None, generated, fqn)); - } - - writer.Outdent().NewLine().Append("}"); - - if (!string.IsNullOrWhiteSpace(cls.Namespace)) - { - writer.Outdent().NewLine().Append("}"); - } - writer.NewLine(); - - - void AppendInterfaceDeclaration(KnownInterfaces knownType) - { - if ((cls.Interfaces & knownType) is 0 | !iKeyed.TryGetValue(knownType, out var iType)) return; - writer.Append(isFirst ? " : " : ", ").Append("global::") - .Append(iType.Namespace).Append('.') .Append(iType.Name); - isFirst = false; - } - - void AppendInterfaceMethods(KnownInterfaces knownType) - { - if ((cls.Interfaces & knownType) is 0 | !iKeyed.TryGetValue(knownType, out var iType)) return; - foreach (var method in iType.Methods) - { - if (SkipMethod(method)) continue; // wonky by nature - left for the caller to implement manually - if (declared.Contains(SignatureKey(method))) continue; // the class implements this itself - generated++; - - writer.NewLine().Append(method.ReturnType).Append(" global::") - .Append(iType.Namespace).Append('.').Append(iType.Name).Append('.').Append(method.Name).Append("("); - bool firstParam = true; - foreach (var p in method.Parameters.Span) - { - if (firstParam) firstParam = false; - else writer.Append(", "); - writer.Append(p.Type).Append(" ").Append(p.Name); - } - writer.Append(')').Indent().NewLine(); - - // async methods route to ExecuteAsync (its own retry policy); everything else uses - // Execute. The state struct is shared regardless of return type - keyed on parameter - // types only - so e.g. ArraySet and ArraySetAsync land on the same _tupleN. The return - // type is Task-stripped so unmapping is keyed on the inner result the retry machinery - // actually sees, not the Task wrapper. - bool isAsync = IsAsync(method.ReturnType); - var simpleResult = StripTask(method.ReturnType); - bool needsMap = NeedsMap(simpleResult); - if (needsMap) allReturns.Add(simpleResult); - int tuple = GetTupleIndex(method.Parameters, needsMap); - writer.Append(isAsync ? "=> ExecuteAsync(new _tuple" : "=> Execute(new _tuple").Append(tuple).Append("("); - firstParam = true; - foreach (var p in method.Parameters.Span) - { - if (firstParam) firstParam = false; - else writer.Append(", "); - writer.Append(p.Name); - } - // the funnels take the state by readonly-ref (see AutoDatabase{Sync|Async}Operation) to avoid - // copying the larger state structs; a lambda only binds to an `in` parameter if it says so, - // hence the explicit modifier (C# 14 allows it without also naming the type) - writer.Append("), static (in state, inner) => inner.").Append(method.Name).Append("("); - for (int i = 0; i < method.Parameters.Length; i++) - { - if (i != 0) writer.Append(", "); - writer.Append("state.Arg").Append(i); - } - writer.Append("));").Outdent().NewLine(); - } - } - - static string GetTupleKey(BasicArray parameters) - { - var keyBuilder = new StringBuilder(); - foreach (var p in parameters.Span) - { - keyBuilder.Append(p.Type).Append('|'); // types-only key; no ref/out on this surface, so type is sufficient - } - - return keyBuilder.ToString(); - } - - bool TupleNeedsMap(BasicArray parameters) => - tupleIndex.TryGetValue(GetTupleKey(parameters), out var found) && found.NeedsMap; - - int GetTupleIndex(BasicArray parameters, bool needsMap) - { - var key = GetTupleKey(parameters); - int index; - if (tupleIndex.TryGetValue(key, out var found)) - { - index = found.Index; - if (needsMap && !found.NeedsMap) - { - found.NeedsMap = true; - tupleIndex[key] = found; - } - } - else - { - index = tupleDefs.Count; - tupleIndex.Add(key, (index, needsMap)); - tupleDefs.Add(parameters); - } - - return index; - } - - const string CommandFlagsType = "StackExchange.Redis.CommandFlags"; - const string RedisKeyType = "StackExchange.Redis.RedisKey"; - const string RedisChannelType = "StackExchange.Redis.RedisChannel"; - // types that carry key(s) internally without "RedisKey" in their name, so the - // substring check below won't catch them; they rely on a Map extension method - const string StreamPositionType = "StackExchange.Redis.StreamPosition"; - // the Execute/ExecuteAsync escape hatch boxes keys/channels inside a loosely-typed - // arg list; these route through a Map extension that unboxes and rewrites matches - const string ScriptArgArrayType = "object[]"; - const string ScriptArgCollectionType = "System.Collections.Generic.ICollection"; - - static bool NeedsMap(string name) => - name.IndexOf(RedisKeyType, StringComparison.Ordinal) >= 0 - || name.IndexOf(RedisChannelType, StringComparison.Ordinal) >= 0 - || name.IndexOf(StreamPositionType, StringComparison.Ordinal) >= 0 - || name == ScriptArgArrayType - || name == ScriptArgCollectionType - || name.IndexOf("ListPopResult", StringComparison.Ordinal) >= 0 - || name.IndexOf("SortedSetPopResult", StringComparison.Ordinal) >= 0 - || name == ScriptArgCollectionType + "?"; - - void AppendTupleTypes() - { - for (int i = 0; i < tupleDefs.Count; i++) - { - var raw = tupleDefs[i]; - bool needsMap = TupleNeedsMap(raw); - var parameters = raw.Span; - - // key/channel-bearing fields are mutable only when something can rewrite them (i.e. the - // owning database is an IRedisArgsMutator); otherwise every field is readonly - // a non-mutator's captured args never change after capture, so the struct itself can be - // readonly - which also guarantees no defensive copies when read through an `in` ref - writer.NewLine().NewLine().Append(cls.IsMutator ? "private struct _tuple" : "private readonly struct _tuple").Append(i).Append("("); - for (int p = 0; p < parameters.Length; p++) - { - if (p != 0) writer.Append(", "); - writer.Append(parameters[p].Type).Append(" arg").Append(p); - } - - // captured args are a plain struct unless the owning database rewrites keys, in which - // case the mapping members - and the interface that carries them - are emitted below. - // A captured CommandFlags is surfaced through IFlaggedRedisArgs regardless, so that a - // funnel holding an opaque TState can still see fire-and-forget; the two compose. - writer.Append(")"); - int flagsArg = -1; - for (int p = 0; p < parameters.Length; p++) - { - // suffix rather than substring: this must not match a nullable, an array, or a - // container *of* flags, none of which the funnel could read as a plain value - if (parameters[p].Type.EndsWith(CommandFlagsType, StringComparison.Ordinal)) - { - flagsArg = p; - break; - } - } - - bool firstBase = true; - if (cls.IsMutator) - { - writer.Append(" : global::StackExchange.Redis.IMappableRedisArgs"); - firstBase = false; - } - - if (flagsArg >= 0) - { - writer.Append(firstBase ? " : " : ", ").Append("global::StackExchange.Redis.IFlaggedRedisArgs"); - firstBase = false; - } - - // a replaying funnel constrains its TState to IDisposable so it can give cloned - // arguments back; every one of its captures implements it, most as a no-op - if (cls.Replays) - { - writer.Append(firstBase ? " : " : ", ").Append("global::System.IDisposable"); - } - writer.NewLine().Append("{").Indent(); - for (int p = 0; p < parameters.Length; p++) - { - var clone = cls.Replays && IsMemoryArg(parameters[p].Type); - writer.NewLine().Append("public ").Append(cls.IsMutator && NeedsMap(parameters[p].Type) ? "" : "readonly ").Append(parameters[p].Type) - .Append(" Arg").Append(p).Append(" = ") - .Append(clone ? "global::StackExchange.Redis.CapturedArgs.Clone(arg" : "arg").Append(p).Append(clone ? ");" : ";"); - } - - if (flagsArg >= 0) - { - // explicit implementation, so it cannot collide with a captured argument's own name; - // readonly only on a mutator's struct, since a readonly struct's members already are - writer.NewLine().Append(cls.IsMutator ? "readonly " : "") - .Append("global::StackExchange.Redis.CommandFlags global::StackExchange.Redis.IFlaggedRedisArgs.Flags => Arg") - .Append(flagsArg).Append(";"); - } - - if (cls.Replays) - { - // emitted even when there is nothing to give back: the funnel's TState constraint - // needs every capture to satisfy it, and an empty one inlines away - writer.NewLine().Append(cls.IsMutator ? "readonly " : "").Append("public void Dispose()").NewLine().Append("{").Indent(); - for (int p = 0; p < parameters.Length; p++) - { - if (!IsMemoryArg(parameters[p].Type)) continue; - writer.NewLine().Append("global::StackExchange.Redis.CapturedArgs.Release(in Arg").Append(p).Append(");"); - } - writer.Outdent().NewLine().Append("}"); - } - - if (!cls.IsMutator) - { - // nothing rewrites keys here, so there is no Map/UnMapper to emit at all - writer.Outdent().NewLine().Append("}"); - continue; - } - - // Map rewrites each scalar key/channel field directly, and defers container or - // loosely-typed fields to a matching Map extension method - writer.NewLine().Append("public void Map(global::StackExchange.Redis.IRedisArgsMutator mutator)") - .NewLine().Append("{").Indent(); - for (int p = 0; p < parameters.Length; p++) - { - if (NeedsMap(parameters[p].Type)) - { - // key/channel-bearing container (or the loosely-typed script arg list); it - // is the library's job to ensure a suitable Map extension method exists - writer.NewLine().Append("Arg").Append(p).Append(" = mutator.Map(Arg").Append(p).Append(");"); - } - } - writer.Outdent().NewLine().Append("}"); - - // tuples with key/channel-bearing results point UnMapper at the shared singleton - // (which knows how to unmap every such result type); the rest return null so the - // Execute helper skips unmapping entirely - and neither path boxes the struct - writer.NewLine().Append("public readonly object? UnMapper => ") - .Append(needsMap ? "_UnMapper.Instance;" : "null;"); - - writer.Outdent().NewLine().Append("}"); - } - - if (cls.IsMutator && allReturns.Count is not 0) - { - // sorted for deterministic (cache-stable) output - var ordered = new List(allReturns); - ordered.Sort(StringComparer.Ordinal); - - writer.NewLine().NewLine().Append("private sealed class _UnMapper").Indent(); - bool firstIface = true; - foreach (var retType in ordered) - { - writer.NewLine().Append(firstIface ? ": " : ", ") - .Append("global::StackExchange.Redis.IRedisArgsResult<").Append(retType).Append(">"); - firstIface = false; - } - writer.Outdent().NewLine().Append("{").Indent(); - writer.NewLine().Append("public static readonly _UnMapper Instance = new();"); - foreach (var retType in ordered) - { - writer.NewLine().NewLine().Append(retType).Append(' ') - .Append("global::StackExchange.Redis.IRedisArgsResult<") - .Append(retType) - .Append(">.UnMap(global::StackExchange.Redis.IRedisArgsMutator mutator, ") - .Append(retType).Append(" value)") - .Indent().NewLine().Append("=> mutator.UnMap(value);").Outdent(); - } - writer.Outdent().NewLine().Append("}"); - } - } - } - writer.NewLine(); - - ctx.AddSource("AutoDatabase.generated.cs", sb.ToString()); - } -} diff --git a/eng/StackExchange.Redis.Build/Diagnostics.cs b/eng/StackExchange.Redis.Build/Diagnostics.cs index 949388f43..06a0828a7 100644 --- a/eng/StackExchange.Redis.Build/Diagnostics.cs +++ b/eng/StackExchange.Redis.Build/Diagnostics.cs @@ -383,32 +383,5 @@ internal static class Diagnostics description: "The RespFragment generator implements [Resp] partial properties of type RespFragment or RespCommand; a declaration it cannot match is skipped, which would otherwise appear only as a missing implementation part.", helpLinkUri: HelpLink("SER351")); - /// - /// An [AutoDatabase(WarnIfIncomplete = true)] type still has members that only throw. - /// - /// - /// - /// A transition tripwire, not a code-quality rule. A type being migrated to a new implementation one - /// command at a time has the rest generated as throwing stubs; that is fine while the work is in - /// progress and not fine in something that ships, and the difference is invisible at a glance - /// because the whole point of the generated half is that nobody writes or reads it. - /// - /// - /// Reported only for Release builds (detected by the absence of the DEBUG preprocessor - /// symbol) so the inner development loop stays quiet, and only where the attribute opts in - the other - /// [AutoDatabase] users generate members that genuinely work by forwarding, and counting those - /// as "not implemented" would be both wrong and deafening. - /// - /// - public static readonly DiagnosticDescriptor AutoDatabaseIncomplete = new( - id: "SER352", - title: "Generated database members are not implemented", - messageFormat: "{0} member(s) of '{1}' are not implemented and will throw at run time", - category: BuildCategory, - defaultSeverity: DiagnosticSeverity.Warning, - isEnabledByDefault: true, - description: "A type generated by [AutoDatabase(WarnIfIncomplete = true)] still has members that only throw; it is mid-transition and should not ship in that state.", - helpLinkUri: HelpLink("SER352")); - private static string HelpLink(string id) => string.Format(HelpLinkFormat, id); } diff --git a/src/StackExchange.Redis/AutoDatabase.cs b/src/StackExchange.Redis/AutoDatabase.cs deleted file mode 100644 index a7bcef4ee..000000000 --- a/src/StackExchange.Redis/AutoDatabase.cs +++ /dev/null @@ -1,294 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Diagnostics; -using System.Diagnostics.CodeAnalysis; -using System.Runtime.CompilerServices; -using System.Threading.Tasks; - -namespace StackExchange.Redis; - -[Conditional("DEBUG")] -[AttributeUsage(AttributeTargets.Class | AttributeTargets.Interface, AllowMultiple = false)] -internal sealed class AutoDatabaseAttribute : Attribute -{ - /// - /// Warn (SER352) on Release builds while any member of the decorated type is still generated - /// rather than implemented by the type itself. - /// - /// - /// For a type that is mid-transition, where the generated members throw rather than forward. Off by - /// default, because the other users of this attribute generate members that genuinely work - counting - /// those as unimplemented would be both wrong and deafening. - /// - public bool WarnIfIncomplete { get; set; } - - /// - /// Whether the owning database can invoke a captured operation more than once, i.e. it replays. - /// - /// - /// A replaying database outlives the call it captured: it re-invokes with the same captured arguments - /// after the caller has long since regained control. Arguments passed as Memory/ReadOnlyMemory - /// still belong to the caller, who is entitled to reuse that buffer once the call returns - so a replay - /// would read whatever is in it by then. When this is set, the generated capture takes a pooled shallow - /// copy of those arguments and the funnel disposes it; when it is not, the arguments are captured as-is, - /// because a single forward cannot outlive the call. - /// - public bool Replays { get; set; } -} - -// Implemented by a generated captured-arguments struct only when the owning auto-database implements -// IRedisArgsMutator - i.e. only when something actually rewrites keys/channels. Everything else captures its -// arguments into a plain readonly struct with no interface at all; a funnel that wants to map opts in by -// constraining its TState to this (which is how KeyPrefixedDatabase will work once it moves onto -// [AutoDatabase]). -// -// Map mutates the struct in place, so a funnel that maps MUST take its state BY VALUE - never by `in`, because -// calling a non-readonly member through an `in` reference silently takes a defensive copy, mutates that, and -// throws it away. Invoke it via RedisArgsMutatorExtensions.MapInPlace, which makes that mistake a compile -// error (CS8329) instead of a silent loss of prefixing. (`ref` on the funnel is not an alternative: the -// generated call site passes a constructed temporary, which `ref` rejects with CS1510, and async funnels -// cannot have by-ref parameters at all.) The by-value copy is not waste here - a mapping funnel needs its own -// mutable copy by definition. Only non-mapping funnels take `in`, and their structs are readonly, so for them -// no defensive copy is even possible. -internal interface IMappableRedisArgs -{ - [Obsolete("Use MapInPlace instead; calling Map directly does nothing when the state is held by `in`.")] - void Map(IRedisArgsMutator mutator); - - object? UnMapper { get; } -} - -// Implemented by a generated captured-arguments struct whenever the captured call has a CommandFlags -// parameter - which is very nearly all of them - so that a funnel holding an opaque TState can still ask the -// one question that changes what it must hand back: was this fire-and-forget? -// -// It existed for the retired capture-and-replay retry transaction, and remains for any funnel with the -// same need. That funnel recorded each call and returned a durable proxy task completed -// later from the attempt, which is wrong for fire-and-forget: a plain RedisTransaction hands back an -// *already-completed* task there (the caller has explicitly declined the reply), so without this the same -// source awaiting the same result returns instantly on one and hangs on the other. The flags are captured -// inside TState and are otherwise invisible to a funnel, which is the whole reason this interface is needed -// rather than the funnel simply reading a parameter. -internal interface IFlaggedRedisArgs -{ - CommandFlags Flags { get; } -} - -// The projections used by the auto-database funnels. There are exactly four real shapes, so each is named -// rather than being expressed generically over the target/return type: that lets the type system encode the -// invariants (sync goes to IDatabase, async goes to IDatabaseAsync and returns a Task) instead of leaving -// nonsense combinations like "sync database, Task result" expressible. -// -// They exist at all (rather than Func<,,>/Action<,>) purely to pass the captured state by readonly-reference: -// some of the generated state structs are chunky (a dozen-plus arguments), and by-value would copy the whole -// thing on every call. Note that a lambda only binds to an `in` parameter if it says so - the generator emits -// `static (in state, inner) => ...` accordingly. -// -// TState is necessarily invariant: it is passed by readonly-ref, and variant type parameters cannot carry -// the struct constraint. TResult is covariant on the sync projection (it is the direct return); it cannot be -// on the async one, because there it appears inside the invariant Task. -internal delegate void AutoDatabaseSyncOperation(in TState state, IDatabase database) - where TState : struct; - -internal delegate TResult AutoDatabaseSyncOperation(in TState state, IDatabase database) - where TState : struct; - -internal delegate Task AutoDatabaseAsyncOperation(in TState state, IDatabaseAsync database) - where TState : struct; - -internal delegate Task AutoDatabaseAsyncOperation(in TState state, IDatabaseAsync database) - where TState : struct; - -internal interface IRedisArgsMutator -{ - RedisKey Map(RedisKey key); - RedisChannel Map(RedisChannel channel); - - RedisKey UnMap(RedisKey key); - RedisChannel UnMap(RedisChannel channel); -} - -internal interface IRedisArgsResult -{ - T UnMap(IRedisArgsMutator mutator, T value); -} - -internal static class RedisArgsMutatorExtensions -{ - // these are used by the generated tuple-types via auto-database: each maps the key-bearing parts - // of an argument through the supplied mutator. They hang off IRedisArgsMutator (rather than the - // argument) so a call site reads consistently with the interface's own MapKey/MapChannel. - public static KeyValuePair Map( - this IRedisArgsMutator mutator, - KeyValuePair value) => - new(mutator.Map(value.Key), value.Value); - - // Funnels should map via this rather than calling state.Map(mutator) directly. Mapping mutates the struct - // in place, so it is only correct on a writable variable; the `ref this` here means that applying it to an - // `in` parameter fails to compile (CS8329) instead of silently taking a defensive copy, mutating that, and - // discarding it - which would leave keys unprefixed, i.e. a wrong-keyspace bug rather than a perf nit. - // (Deliberately not named Map: an extension cannot shadow the instance member, so the guard would be lost.) - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static void MapInPlace(this ref TState state, IRedisArgsMutator mutator) - where TState : struct, IMappableRedisArgs -#pragma warning disable CS0618 // the one sanctioned call site: `state` is a genuine `ref`, so this mutates in place - => state.Map(mutator); -#pragma warning restore CS0618 - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static TResult UnMap(this IRedisArgsMutator mutator, in TState state, TResult result) - where TState : struct, IMappableRedisArgs - { - // note JIT will elide these tests using per-TResult value-type rules - if (typeof(TResult) == typeof(RedisKey)) - { - var tmp = mutator.UnMap(Unsafe.As(ref result)); - return Unsafe.As(ref tmp); - } - - if (typeof(TResult) == typeof(RedisChannel)) - { - var tmp = mutator.UnMap(Unsafe.As(ref result)); - return Unsafe.As(ref tmp); - } - - if (typeof(TResult) == typeof(RedisValue)) - { - return result; // never mapped - } - - return state.UnMapper is IRedisArgsResult unmap ? unmap.UnMap(mutator, result) : result; - } - - [return: NotNullIfNotNull("keys")] - public static RedisKey[]? Map(this IRedisArgsMutator mutator, RedisKey[]? keys) - { - if (keys is null || keys.Length is 0) return keys; - var arr = new RedisKey[keys.Length]; - for (int i = 0; i < arr.Length; i++) - { - arr[i] = mutator.Map(keys[i]); - } - return arr; - } - - // ScriptEvaluateResp's keys parameter - every element is a key (unlike RedisKeyOrValue below), - // mirroring the RedisKey[]? overload above but for the ReadOnlyMemory shape. - public static ReadOnlyMemory Map(this IRedisArgsMutator mutator, ReadOnlyMemory keys) - { - if (keys.Length is 0) return keys; - var arr = new RedisKey[keys.Length]; - var span = keys.Span; - for (int i = 0; i < arr.Length; i++) - { - arr[i] = mutator.Map(span[i]); - } - return arr; - } - - // the ExecuteResp escape hatch mixes keys and values in one collection (mirroring KeyPrefixed.ToInner); - // rewrite only the key-shaped entries, copying only when there is something to rewrite so the common - // all-values call allocates nothing. - public static ReadOnlyMemory Map(this IRedisArgsMutator mutator, ReadOnlyMemory args) - { - if (args.Length is 0) return args; - var span = args.Span; - RedisKeyOrValue[]? copy = null; - for (int i = 0; i < span.Length; i++) - { - if (span[i].IsKey) - { - (copy ??= span.ToArray())[i] = RedisKeyOrValue.FromKey(mutator.Map(span[i].Key)); - } - } - return copy ?? args; - } - - [return: NotNullIfNotNull("pairs")] - public static KeyValuePair[]? Map( - this IRedisArgsMutator mutator, - KeyValuePair[]? pairs) - { - if (pairs is null || pairs.Length is 0) return pairs; - var arr = new KeyValuePair[pairs.Length]; - for (int i = 0; i < arr.Length; i++) - { - ref readonly KeyValuePair pair = ref pairs[i]; - arr[i] = new(mutator.Map(pair.Key), pair.Value); - } - return arr; - } - - public static StreamPosition Map(this IRedisArgsMutator mutator, StreamPosition value) => - new(mutator.Map(value.Key), value.Position); - - [return: NotNullIfNotNull("positions")] - public static StreamPosition[]? Map(this IRedisArgsMutator mutator, StreamPosition[]? positions) - { - if (positions is null || positions.Length is 0) return positions; - var arr = new StreamPosition[positions.Length]; - for (int i = 0; i < arr.Length; i++) - { - ref readonly StreamPosition position = ref positions[i]; - arr[i] = new(mutator.Map(position.Key), position.Position); - } - return arr; - } - - // the Execute/ExecuteAsync escape hatch takes a loosely-typed arg list in which any element may - // be a boxed RedisKey or RedisChannel; these rewrite just those entries (mirroring - // KeyPrefixed.ToInner), copying only when there is something to rewrite so the common call allocates nothing. - [return: NotNullIfNotNull("args")] - public static object[]? Map(this IRedisArgsMutator mutator, object[]? args) - { - if (args is null || args.Length is 0) return args; - object[]? copy = null; - for (int i = 0; i < args.Length; i++) - { - if (args[i] is RedisKey key) - { - (copy ??= (object[])args.Clone())[i] = mutator.Map(key); - } - else if (args[i] is RedisChannel channel) - { - (copy ??= (object[])args.Clone())[i] = mutator.Map(channel); - } - } - return copy ?? args; - } - - [return: NotNullIfNotNull("args")] - public static ICollection? Map(this IRedisArgsMutator mutator, ICollection? args) - { - if (args is null || args.Count is 0) return args; - bool any = false; - foreach (var arg in args) - { - if (arg is RedisKey or RedisChannel) - { - any = true; - break; - } - } - if (!any) return args; - - var copy = new object[args.Count]; - int i = 0; - foreach (var arg in args) - { - copy[i++] = arg switch - { - RedisKey key => mutator.Map(key), - RedisChannel channel => mutator.Map(channel), - _ => arg, - }; - } - return copy; - } - - public static SortedSetPopResult UnMap(this IRedisArgsMutator mutator, SortedSetPopResult value) => - value.IsNull ? SortedSetPopResult.Null : new(mutator.Map(value.Key), value.Entries); - - public static ListPopResult UnMap(this IRedisArgsMutator mutator, ListPopResult value) => - value.IsNull ? ListPopResult.Null : new(mutator.Map(value.Key), value.Values); -} diff --git a/src/StackExchange.Redis/Availability/RespRetryExecutor.cs b/src/StackExchange.Redis/Availability/RespRetryExecutor.cs index 2cd23b1ff..c9f2fbae2 100644 --- a/src/StackExchange.Redis/Availability/RespRetryExecutor.cs +++ b/src/StackExchange.Redis/Availability/RespRetryExecutor.cs @@ -54,8 +54,8 @@ internal RespRetryExecutor(RespExecutorBase inner, RetryPolicy policy, FuncThe policy in force; read by the tests, which assert how a policy was resolved. internal RetryPolicy Policy => _controller.Policy; - /// The controller, shared with so a transaction created from a - /// retrying database applies exactly the policy the database does. + /// The controller, shared with so a transaction + /// created from a retrying database applies exactly the policy the database does. internal RetryController Controller => _controller; /// The executor underneath, so a second WithRetry can refuse rather than nest. diff --git a/src/StackExchange.Redis/Availability/RespRetryTransactionExecutor.cs b/src/StackExchange.Redis/Availability/RespRetryTransactionExecutor.cs new file mode 100644 index 000000000..4236ca564 --- /dev/null +++ b/src/StackExchange.Redis/Availability/RespRetryTransactionExecutor.cs @@ -0,0 +1,631 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using StackExchange.Redis.Protocol; + +namespace StackExchange.Redis.Availability; + +/// +/// A transaction that replays as a unit: the frame-level successor to the capture-and-replay +/// RetryTransaction. +/// +/// +/// +/// What is stored is the rendered request, not the method call. Where RetryTransaction +/// recorded every builder invocation into a generated state struct and replayed it against a fresh +/// , this accumulates what the command surface already produced - the rendered +/// , its handler, and the conditions in semantic form - and each attempt feeds +/// them into a fresh, one-shot . Same unit, same loop, no generator. +/// +/// +/// Why replay cannot reuse the attempt's operations. Once a run is written, each queued operation +/// is a pooled, single-completion source enrolled with that connection: the connection's own teardown +/// faults it, and a second enrollment while the first may still touch it is a use-after-release on the +/// pool. So the operations enrolled per attempt are the attempt's own, and the caller holds durable +/// proxies that settle only when the unit has finished for good - exactly the shape +/// RetryTransaction used, one layer down. +/// +/// +/// Why replaying the unit is safe at all: the WATCH constraints live in the conditions, not the +/// queue, and each attempt re-issues them on its own connection - so a replay re-adjudicates against +/// fresh state rather than running with a dead connection's guard silently gone. The queued commands +/// run only if the re-issued conditions hold and the EXEC commits. An EXEC that was +/// written and never answered is NOT replayed blindly: the fault carries the queue's aggregate retry +/// category, so the shared policy gates the whole transaction exactly like one command. +/// +/// +internal sealed class RespRetryTransactionExecutor : RespExecutorBase, ITransactionExecutor +{ + private readonly RespExecutorBase _inner; + private readonly RespContext _context; + private readonly RetryController _controller; + private readonly Func? _failoverSource; + private readonly object _sync = new(); + + private List? _recorded; + private List? _conditions; + private int _executed; + private volatile bool _watchConflict; + + /// Create a retrying transaction over an executor. + /// The executor underneath the retry link; each attempt resolves through it. + /// The context conditions render through; it must NOT carry the retry link. + /// The shared retry controller, from the database's RespRetryExecutor. + internal RespRetryTransactionExecutor(RespExecutorBase inner, RespContext context, RetryController controller) + { + _inner = inner ?? throw new ArgumentNullException(nameof(inner)); + _context = context ?? throw new ArgumentNullException(nameof(context)); + _controller = controller ?? throw new ArgumentNullException(nameof(controller)); + _failoverSource = inner.GetFailoverSource(); + } + + /// + internal override bool Accumulates => true; + + /// + internal override bool Transactional => true; + + /// + public override int Database => _inner.Database; + + /// + internal override IConnectionMultiplexer? ExposedMultiplexer => _inner.ExposedMultiplexer; + + /// + internal override RespExecutorBase? ResolveFor(in RedisKey key, RedisCommand command, CommandFlags flags) + => _inner.ResolveFor(in key, command, flags); + + /// + /// Yes: the recorded request is this executor's own reference (or copy) by the time this returns. + internal override bool CopiesRequestOnSend => true; + + /// + public bool WasWatchConflict => _watchConflict; + + /// + public override RespPayload Send(in RespRequest request) + => throw new NotSupportedException("A queued command cannot be waited on before the transaction is executed."); + + /// + /// + /// The payload shape: the recorded command owns no handler, and the final attempt's payload is handed + /// to the caller intact rather than parsed - the same contract as the plain executor's untyped path. + /// The request is NOT retained here: by that path's convention the caller holds its reference until + /// the returned task completes, which is exactly the final attempt. + /// + public override ValueTask SendAsync(RespRequest request, CancellationToken cancellationToken = default) + { + if ((request.Flags & CommandFlags.FireAndForget) != 0) + { + Record(new RecordedPayload(request, ownsRequest: false, proxy: null, cancellationToken)); + return default; + } + + var proxy = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Record(new RecordedPayload(request, ownsRequest: false, proxy, cancellationToken)); + return new ValueTask(proxy.Task); + } + + /// + internal override ValueTask SendTypedAsync( + RespRequest request, IRespHandler handler, CancellationToken cancellationToken) + { + // the same special case as the plain executor: a payload-typed result goes the untyped route + if (typeof(TResult) == typeof(RespPayload)) return RespExecutor.AwaitUncached(this, request, handler, cancellationToken); + + // ToOwned: a borrowed request is copied, an owned one is kept - this path's convention + // (CopiesRequestOnSend) makes the incoming reference ours to keep, and Release gives it back + var owned = request.ToOwned(); + + if ((owned.Flags & CommandFlags.FireAndForget) != 0) + { + Record(new Recorded(owned, ownsRequest: true, handler, proxy: null, _inner, cancellationToken)); + + // the one shared completed-default task, as the plain transaction hands out for the same case + return new ValueTask(CompletedTask.Default(null)!); + } + + var proxy = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Record(new Recorded(owned, ownsRequest: true, handler, proxy, _inner, cancellationToken)); + return new ValueTask(proxy.Task); + } + + /// + internal override ValueTask SendVoidAsync(RespRequest request, IRespHandler handler, CancellationToken cancellationToken) + { + var pending = SendTypedAsync(request, handler, cancellationToken); + if (pending.IsCompletedSuccessfully) + { + _ = pending.Result; + return default; + } + + return new ValueTask(pending.AsTask()); + } + + private void Record(IRecorded recorded) + { + lock (_sync) + { + if (_executed != 0) throw new InvalidOperationException("This transaction has already been executed."); + (_recorded ??= []).Add(recorded); + } + } + + /// + public void AddCondition(Condition condition, Action? onVerdict = null) + { + if (condition is null) throw new ArgumentNullException(nameof(condition)); + lock (_sync) + { + if (_executed != 0) throw new InvalidOperationException("Operations cannot be added after the transaction has been executed."); + (_conditions ??= []).Add(new RecordedCondition(condition, onVerdict)); + } + } + + /// + /// The fold the retry policy gates the whole unit with; see the plain executor's twin. + public CommandFlags AggregateRetryCategory + { + get + { + var result = CommandFlags.None; + lock (_sync) + { + if (_recorded is { } recorded) + { + foreach (var op in recorded) + { + var category = op.Flags & CommandFlagsInternal.MaskRetryCategory; + if (category > result) result = category; + } + } + } + + return result; + } + } + + /// + public async Task ExecuteAsync(CommandFlags flags = CommandFlags.None) + { + List? recorded; + List? conditions; + lock (_sync) + { + if (_executed != 0) throw new InvalidOperationException("This transaction has already been executed."); + _executed = 1; + + // taken under the lock, so the loops below run over lists nothing can still be adding to + recorded = _recorded; + conditions = _conditions; + _recorded = null; + _conditions = null; + } + + try + { + return await ExecuteCoreAsync(flags, recorded ?? [], conditions).ConfigureAwait(false); + } + finally + { + if (recorded is not null) + { + foreach (var op in recorded) op.Release(); + } + } + } + + private async Task ExecuteCoreAsync(CommandFlags flags, List recorded, List? conditions) + { + var attemptCount = 0; + + // captured BEFORE the first attempt - otherwise a failover between a failed attempt and + // re-reading the token could be missed (same reasoning as RespRetryExecutor) + var failover = _controller.TracksFailover ? (_failoverSource?.Invoke() ?? CancellationToken.None) : CancellationToken.None; + + // watch contention gets its own budget; only meaningful when there are conditions to WATCH + var watchAttempt = 0; + var maxWatchAttempts = conditions is null ? 1 : _controller.MaxWatchConflictAttempts; + while (true) + { + // a fresh, one-shot unit per attempt: the conditions re-render and re-WATCH on whatever + // connection this attempt lands on, and the queued frames re-enroll as that attempt's own + // operations - the durable proxies the caller holds are settled only at the end + var attempt = new RespTransactionExecutor(_inner, _context, reusable: false); + if (conditions is not null) + { + foreach (var condition in conditions) condition.Replay(attempt); + } + + foreach (var op in recorded) op.Replay(attempt); + + // the aggregate category rides on the EXEC's flags, so the fault it reports carries it and + // the shared policy gates the whole transaction exactly like one command + var effectiveFlags = (flags & ~CommandFlagsInternal.MaskRetryCategory) | attempt.AggregateRetryCategory; + + try + { + var committed = await attempt.ExecuteAsync(effectiveFlags).ConfigureAwait(false); + _watchConflict = attempt.WasWatchConflict; + + // the server rejected an EXEC we really did issue because a watched key changed under us: + // contention rather than a fault, so it has its own budget, waits its own delay, and the + // side-effect category does not apply - nothing happened + if (!committed && _watchConflict && ++watchAttempt < maxWatchAttempts) + { + DiscardAttempt(recorded, conditions); + await _controller.WatchConflictDelayAsync().ConfigureAwait(false); + continue; + } + + // clean completion (committed, or electively aborted): forward the per-attempt outcomes + // onto the durable proxies - settled before we return, unless the caller declined the + // EXEC's own reply, in which case waiting is the opposite of what they asked + if (conditions is not null) + { + foreach (var condition in conditions) condition.Forward(); + } + + List? settling = null; + foreach (var op in recorded) + { + if (op.ForwardOutcome() is { } pending) (settling ??= []).Add(pending); + } + + if (settling is not null && (effectiveFlags & CommandFlags.FireAndForget) == 0) + { + foreach (var pending in settling) + { + await pending.ConfigureAwait(false); // never faults: the forwarders catch and settle the proxy + } + } + + return committed; + } + catch (Exception ex) + { + if (_controller.CanRetry(++attemptCount, ex, ref failover, out var delay)) + { + DiscardAttempt(recorded, conditions); + await _controller.FailoverOrDelayAsync(delay).ConfigureAwait(false); + continue; + } + + // out of road: fault the durable proxies and surface the failure to the caller + foreach (var op in recorded) op.Fault(ex); + throw; + } + } + } + + /// Consume a discarded attempt's results and reset for the next. + private static void DiscardAttempt(List recorded, List? conditions) + { + foreach (var op in recorded) op.Drain(); + if (conditions is not null) + { + foreach (var condition in conditions) condition.Reset(); + } + } + + // ---- recorded work ----------------------------------------------------------------------------- + private interface IRecorded + { + /// The recorded request's flags, for the aggregate retry category. + CommandFlags Flags { get; } + + /// Queue the recorded request onto this attempt's unit. + void Replay(RespTransactionExecutor attempt); + + /// + /// Copy this attempt's outcome onto the durable proxy; the returned task is the settling still in + /// flight (a fire-and-forget EXEC completes before the replies land), or null when it is done. + /// + Task? ForwardOutcome(); + + /// Consume a discarded attempt's result, so a pooled source is released and a fault is observed. + void Drain(); + + /// Fault the durable proxy; the unit is out of attempts. + void Fault(Exception ex); + + /// Give back the recorded request, once, after the unit has finished with it for good. + void Release(); + } + + /// A recorded command whose caller wants a parsed . + private sealed class Recorded : IRecorded + { + private readonly RespRequest _request; + private readonly bool _ownsRequest; + private readonly IRespHandler _handler; + private readonly TaskCompletionSource? _proxy; // null: fire-and-forget declined the reply + private readonly RespExecutorBase _authSource; + private readonly CancellationToken _cancellationToken; + private ValueTask _attempt; + private bool _hasAttempt; + + public Recorded( + in RespRequest request, + bool ownsRequest, + IRespHandler handler, + TaskCompletionSource? proxy, + RespExecutorBase authSource, + CancellationToken cancellationToken) + { + _request = request; + _ownsRequest = ownsRequest; + _handler = handler; + _proxy = proxy; + _authSource = authSource; + _cancellationToken = cancellationToken; + } + + public CommandFlags Flags => _request.Flags; + + public void Replay(RespTransactionExecutor attempt) + { + _attempt = attempt.SendAsync(_request, _cancellationToken); + _hasAttempt = true; + } + + public Task? ForwardOutcome() + { + if (_proxy is null) + { + Drain(); // fire-and-forget: nothing to carry, but a pooled source still has to be consumed + return null; + } + + if (!_hasAttempt) return null; // loud failure below would hide the real bug; Replay always precedes + var attempt = _attempt; + _attempt = default; + _hasAttempt = false; + if (attempt.IsCompleted) + { + Forward(attempt); + return null; + } + + return ForwardAsync(attempt); + } + + private void Forward(ValueTask attempt) + { + RespPayload? payload = null; + try + { + payload = attempt.GetAwaiter().GetResult(); + _proxy!.TrySetResult(RespExecutor.Parse(_handler, payload)); + } + catch (OperationCanceledException) + { + _proxy!.TrySetCanceled(); + } + catch (RedisServerException ex) when (RespExecutor.AuthFault(_authSource, ex) is { } authFault) + { + _proxy!.TrySetException(authFault); + } + catch (Exception ex) + { + _proxy!.TrySetException(ex); + } + finally + { + payload?.Release(); + } + } + + private async Task ForwardAsync(ValueTask attempt) + { + RespPayload? payload = null; + try + { + payload = await attempt.ConfigureAwait(false); + _proxy!.TrySetResult(RespExecutor.Parse(_handler, payload)); + } + catch (OperationCanceledException) + { + _proxy!.TrySetCanceled(); + } + catch (RedisServerException ex) when (RespExecutor.AuthFault(_authSource, ex) is { } authFault) + { + _proxy!.TrySetException(authFault); + } + catch (Exception ex) + { + _proxy!.TrySetException(ex); + } + finally + { + payload?.Release(); + } + } + + public void Drain() + { + if (!_hasAttempt) return; + var attempt = _attempt; + _attempt = default; + _hasAttempt = false; + DrainCore(attempt); + } + + public void Fault(Exception ex) + { + Drain(); + _proxy?.TrySetException(ex); + } + + public void Release() + { + if (_ownsRequest) _request.Dispose(); + } + } + + /// A recorded command whose caller wants the raw payload, handed over intact. + private sealed class RecordedPayload : IRecorded + { + private readonly RespRequest _request; + private readonly bool _ownsRequest; + private readonly TaskCompletionSource? _proxy; + private readonly CancellationToken _cancellationToken; + private ValueTask _attempt; + private bool _hasAttempt; + + public RecordedPayload(in RespRequest request, bool ownsRequest, TaskCompletionSource? proxy, CancellationToken cancellationToken) + { + _request = request; + _ownsRequest = ownsRequest; + _proxy = proxy; + _cancellationToken = cancellationToken; + } + + public CommandFlags Flags => _request.Flags; + + public void Replay(RespTransactionExecutor attempt) + { + _attempt = attempt.SendAsync(_request, _cancellationToken); + _hasAttempt = true; + } + + public Task? ForwardOutcome() + { + if (_proxy is null) + { + Drain(); + return null; + } + + if (!_hasAttempt) return null; + var attempt = _attempt; + _attempt = default; + _hasAttempt = false; + if (attempt.IsCompleted) + { + Forward(attempt); + return null; + } + + return ForwardAsync(attempt); + } + + // ownership transfers to the caller: the payload is NOT released here + private void Forward(ValueTask attempt) + { + try + { + _proxy!.TrySetResult(attempt.GetAwaiter().GetResult()); + } + catch (OperationCanceledException) + { + _proxy!.TrySetCanceled(); + } + catch (Exception ex) + { + _proxy!.TrySetException(ex); + } + } + + private async Task ForwardAsync(ValueTask attempt) + { + try + { + _proxy!.TrySetResult(await attempt.ConfigureAwait(false)); + } + catch (OperationCanceledException) + { + _proxy!.TrySetCanceled(); + } + catch (Exception ex) + { + _proxy!.TrySetException(ex); + } + } + + public void Drain() + { + if (!_hasAttempt) return; + var attempt = _attempt; + _attempt = default; + _hasAttempt = false; + DrainCore(attempt); + } + + public void Fault(Exception ex) + { + Drain(); + _proxy?.TrySetException(ex); + } + + public void Release() + { + if (_ownsRequest) _request.Dispose(); + } + } + + /// Consume a discarded attempt's result: release any payload, observe any fault. + private static void DrainCore(ValueTask attempt) + { + if (attempt.IsCompleted) + { + try + { + attempt.GetAwaiter().GetResult()?.Release(); + } + catch + { + // a discarded attempt; the outcome that matters is the retry decision already made + } + + return; + } + + _ = Awaited(attempt); + + static async Task Awaited(ValueTask pending) + { + try + { + (await pending.ConfigureAwait(false))?.Release(); + } + catch + { + // observed; see above + } + } + } + + /// A condition, re-issued per attempt, whose durable verdict sink hears only the final attempt. + private sealed class RecordedCondition + { + private readonly Condition _condition; + private readonly Action? _sink; + private bool _hasVerdict; + private bool _verdict; + + public RecordedCondition(Condition condition, Action? sink) + { + _condition = condition; + _sink = sink; + } + + public void Replay(RespTransactionExecutor attempt) + => attempt.AddCondition(_condition, OnVerdict); + + private void OnVerdict(bool satisfied) + { + _verdict = satisfied; + _hasVerdict = true; + } + + public void Reset() => _hasVerdict = false; + + public void Forward() + { + if (_hasVerdict) _sink?.Invoke(_verdict); + } + } +} diff --git a/src/StackExchange.Redis/Availability/RetryTransaction.cs b/src/StackExchange.Redis/Availability/RetryTransaction.cs deleted file mode 100644 index 351bafb8e..000000000 --- a/src/StackExchange.Redis/Availability/RetryTransaction.cs +++ /dev/null @@ -1,537 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Runtime.CompilerServices; -using System.Threading; -using System.Threading.Tasks; -using StackExchange.Redis.Interfaces; - -namespace StackExchange.Redis.Availability; - -// A retryable transaction. Unlike RespRetryExecutor - which replays a single rendered frame each attempt - -// a transaction is built up across many builder calls plus an Execute, so we *record* each captured call -// (the AutoDatabase (state, projection) pair) and hand the caller a durable, still-incomplete proxy task. -// Only ExecuteAsync actually runs anything: for each attempt it spins up a fresh, one-shot inner transaction -// against the underlying database (which, for a multi-group connection, resolves the currently-active member, -// so a retry after failover lands on the new member), replays every recorded operation and constraint onto -// it, and awaits it. On a clean execution the per-attempt results are forwarded onto the durable proxies; on -// a transient fault the whole attempt is discarded and replayed; on terminal failure the proxies are faulted. -// -// ITransaction (rather than just ITransactionAsync) so that callers written against the long-standing -// interface can move to a retrying database without rewriting their transaction code; the synchronous -// Execute is sync-over-async (see below). Note that ITransaction : IBatch : IDatabaseAsync, so the only -// members this adds over ITransactionAsync are the two Execute overloads. -// Replays: the recorded operations are re-run as a unit on every attempt, and a capture is held from the -// moment it is recorded until Execute - longer than a single dispatch holds one - so the captured arguments must -// be copies of the caller's memory rather than the caller's own buffer. They are released in ExecuteAsync, -// which is the only point that knows an operation will not be replayed again; a transaction that is never -// executed keeps its captures, and nothing here could know otherwise. -[AutoDatabase(Replays = true)] -internal sealed partial class RetryTransaction : IDatabaseAsync, ITransaction -{ - /// - public RespDatabaseContext Context => new(GetContextCore()); - // Note: the *command* surface is async-only - retrying is inherently - // delay-ish; only the terminal Execute is offered synchronously. - private readonly IDatabaseAsync _source; - private readonly RetryController _controller; - - /// Refuse to build a context, saying why. - /// - /// Still refused, where a retrying database's is not. A retrying database replays one - /// operation, so decorating its executor gives the context surface exactly the same behaviour. A - /// retrying transaction replays the whole recorded unit - conditions, queued operations and - /// EXEC together - and a retry executor knows nothing of that: it would re-send individual - /// frames inside a transaction, which is not a retry of anything the caller asked for. Retry arrives - /// here when the context surface has transactions, and not before. - /// - internal static RespContext GetContextCore() - => throw new NotImplementedException( - "The context surface does not support retrying transactions; a transaction is replayed as a " - + "unit, which a per-frame retry executor cannot express."); - - /// - RespContext IRespTarget.Context => GetContextCore(); - - // not readonly, and null until something is recorded: ExecuteAsync takes it and clears it, so the - // replay loop below runs over a list nothing else can still be adding to. A caller doing something - // strange - recording from another thread while executing - then gets an empty transaction rather than - // a collection mutated mid-enumeration, and the captures it recorded are its own to lose. - private List? _ops; - private List? _conditions; - private int _executed; - private volatile bool _watchConflict; - - /// - // 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; - } - - public int Database => _source.Database; - - public IConnectionMultiplexer Multiplexer => _source.Multiplexer; - - /// - public override string ToString() => "retry-transaction: " + _source; - - // the generated explicit interface implementations funnel every builder call through these two - // overloads, capturing the arguments in a generated state struct plus a cacheable static projection - // (no per-call closure). Here we simply *record* them and return a durable proxy task. - private Task ExecuteAsync(in TState state, AutoDatabaseAsyncOperation operation) - where TState : struct, IDisposable - { - CheckNotExecuted(); - var op = new RecordedOp(state, operation, IsFireAndForget(in state)); - (_ops ??= new List()).Add(op); - return op.Proxy; - } - - private Task ExecuteAsync(in TState state, AutoDatabaseAsyncOperation operation) - where TState : struct, IDisposable - { - CheckNotExecuted(); - var op = new RecordedVoidOp(state, operation, IsFireAndForget(in state)); - (_ops ??= new List()).Add(op); - return op.Proxy; - } - - // A fire-and-forget operation must hand back an already-completed task, because that is what a plain - // RedisTransaction does (see its ExecuteAsync: F+F never gets a TaskCompletionSource at all). Without this - // the durable proxy stays incomplete until Execute, so identical caller code returns instantly on a plain - // transaction and blocks for good on a retrying one - which defeats this class's stated goal of letting - // ITransaction code move onto a retrying database unchanged. - // - // The op is still recorded and replayed like any other: fire-and-forget declines the *reply*, not the - // command. What it does not get is a proxy - see RecordedOp - so forwarding and faulting both become - // no-ops for it, and a discarded attempt can never fault a result the caller was told not to expect. - // - // The flags are captured inside TState, so this is the one thing the generated struct is asked to expose - // (IFlaggedRedisArgs). Testing a generic value against an interface boxes in IL, and this sits on every - // queued operation - but value-type generics are not shared, so the JIT knows the exact TState and - // compiles the whole thing to a field load with no allocator call at all (11 bytes of machine code). - // - // Measured on net10.0 rather than reasoned about, because none of it is obvious - bytes per call: - // - // `is` pattern, tier-0 ......... 24 `is` pattern, optimized ....... 0 - // cast instead of `is` ......... 24 ...with AggressiveOptimization 0 - // - // Two findings worth keeping. Rewriting the `is` pattern as a cast does *not* avoid the box - both - // spellings emit the same one - so the attribute below is the fix rather than a different expression. - // And a TState *without* the interface costs nothing even in tier-0: the test folds to a constant and - // the dead box is removed, so the states that captured no flags never pay for this at all. - - // skips tiering, so the above is free from the first call rather than only after tier-1 promotion -#if NET - [MethodImpl(MethodImplOptions.AggressiveOptimization)] -#endif - private static bool IsFireAndForget(in TState state) - where TState : struct - => state is IFlaggedRedisArgs flagged && (flagged.Flags & CommandFlags.FireAndForget) != 0; - - public ConditionResult AddCondition(Condition condition) - { - if (condition is null) throw new ArgumentNullException(nameof(condition)); - CheckNotExecuted(); - var recorded = new RecordedCondition(condition); - (_conditions ??= new List()).Add(recorded); - return recorded.Result; - } - - public async Task ExecuteAsync(CommandFlags flags = CommandFlags.None) - { - if (Interlocked.CompareExchange(ref _executed, 1, 0) != 0) - { - throw new InvalidOperationException("This transaction has already been executed"); - } - - // take the recordings off the instance before replaying any of them, so the loops below run over - // lists nothing else can still be adding to; a caller recording from another thread mid-execution - // then gets left out rather than mutating a collection under enumeration. The operations' captured - // arguments are released once we are done with them, however that turns out - though a transaction - // that is never executed keeps its captures, since nothing here could know. - var ops = _ops; - var conditions = _conditions; - _ops = null; - _conditions = null; - try - { - return await ExecuteAsync(flags, ops ?? EmptyOps, conditions).ConfigureAwait(false); - } - finally - { - if (ops is not null) - { - foreach (var op in ops) op.Release(); - } - } - } - - private static readonly List EmptyOps = new(); - - private async Task ExecuteAsync(CommandFlags flags, List ops, List? conditions) - { - int attempt = 0; - // capture the next-failover token *before* the first attempt - otherwise a failover between a - // failed attempt and re-reading the token could be missed - CancellationToken failover = _controller.TracksFailover ? _source.GetNextFailover() : CancellationToken.None; - - // 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) - { - // no async-state: a retrying transaction 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 - if (conditions is not null) - { - foreach (var c in conditions) c.Replay(inner); - } - foreach (var op in ops) op.Replay(inner); - - // inject the aggregate retry category onto the EXEC flags so the resulting fault carries it, and - // the shared RetryPolicy/FaultContext logic gates the whole transaction exactly like one command - var category = inner is IInternalTransaction it ? it.GetAggregateRetryCategory() : CommandFlags.CommandRetryNever; - var effectiveFlags = (flags & ~CommandFlagsInternal.MaskRetryCategory) | category; - - 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 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(); - } - List? settling = null; - foreach (var op in ops) - { - if (op.ForwardSuccess() is { } pending) (settling ??= new()).Add(pending); - } - - // SETTLED before we return, not merely scheduled - unless the caller declined the EXEC's own - // reply, in which case there is nothing to wait for and waiting would be the opposite of what - // they asked. Faults are swallowed here on purpose: they belong on the per-command proxy, - // where the caller reads them, and must not fail the transaction as a whole. - if (settling is not null && (effectiveFlags & CommandFlags.FireAndForget) == 0) - { - foreach (var pending in settling) - { - try - { - await pending.ConfigureAwait(false); - } - catch - { - // carried by the proxy; see above - } - } - } - - return committed; - } - catch (Exception ex) - { - if (_controller.CanRetry(++attempt, ex, ref failover, out var delay)) - { - // discard this attempt; observe its faulted per-attempt tasks so they don't surface as - // unobserved, then wait / failover and replay from the recorded snapshot - foreach (var op in ops) op.Observe(); - await _controller.FailoverOrDelayAsync(delay).ConfigureAwait(false); - continue; - } - - // out of road: fault the durable proxies and surface the failure to the caller - foreach (var op in ops) op.Fault(ex); - throw; - } - } - } - - // ---- the synchronous ITransaction surface ------------------------------------------------------- - // Both are explicit implementations: they exist purely so that callers written against ITransaction can - // move onto a retrying database, so they should be reachable only *through* that interface - internal - // code (which has ExecuteAsync) has no business blocking, and this also keeps Execute() and - // Execute(flags) from competing in overload resolution on the concrete type. - - /// - // The IBatch shape: enqueue, and don't ask about the outcome. Mirrors RedisTransaction by mapping onto - // fire-and-forget - which for a *retrying* transaction means there is nothing for the retry machinery to - // observe (no reply is requested, so no fault is ever reported) and it collapses to a single attempt: - // the trade-off fire-and-forget always makes, here extending to the retries as well. - void IBatch.Execute() => ((ITransaction)this).Execute(CommandFlags.FireAndForget); - - /// - // Sync-over-async, deliberately: retrying is inherently delay-ish, which is why the *command* surface - // here is async-only - but a caller written against ITransaction has no async alternative for the - // terminal Execute, and refusing it would mean such code cannot adopt a retrying database at all. - // Blocking here is safe against a captured synchronization context: every await in ExecuteAsync (and in - // RetryController) is ConfigureAwait(false), and the per-operation proxies complete asynchronously. - // - // The wait is *not* bounded by the multiplexer's SyncTimeout - the whole point of a retry is to outlast - // the fault, so the delays between attempts are expected to exceed it. Each individual attempt is still - // bounded by the inner connection's own timeout, so the total remains bounded by the attempt budget. - // GetResult (rather than Wait) so the original exception surfaces, not an AggregateException wrapping it. - bool ITransaction.Execute(CommandFlags flags) => ExecuteAsync(flags).GetAwaiter().GetResult(); - - private void CheckNotExecuted() - { - if (Volatile.Read(ref _executed) != 0) - { - throw new InvalidOperationException("Operations cannot be added after the transaction has been executed"); - } - } - - // ---- recorded work ------------------------------------------------------------------------------- - private interface IRecordedOp - { - void Replay(IDatabaseAsync inner); - - // Copy this attempt's outcome onto the durable proxy, and say whether that had to be DEFERRED: the - // returned task is the attempt whose completion will settle the proxy, or null when it already has. - // - // A *replied* EXEC populates every queued operation's result before its own task completes - // (TransactionProcessor does both while processing that one reply), so the outcome is normally - // available inline. Two cases are not: - // - // - a fire-and-forget EXEC completes as soon as it has been written, with the replies still in - // flight. Waiting there would block for a round trip the caller explicitly declined, which is - // what F+F means and what a plain RedisTransaction does in the same situation - so the caller - // ignores the returned task; - // - the NEW core sets those results inline too, but publishes each caller-facing task through a - // queued continuation, deliberately, so that a reply never runs user code on the IO thread. The - // result is there and the Task has not caught up, which is a distinction no caller should be able - // to see: `await ExecuteAsync()` has always meant the per-command tasks are settled. So the - // caller waits on what is returned, unless the EXEC itself was declined. - Task? ForwardSuccess(); - void Fault(Exception ex); - void Observe(); - - /// - /// Give back anything the captured arguments own. Called once, after the transaction has finished - /// with this operation for good - not per attempt, since every attempt replays the same capture. - /// - void Release(); - } - - private sealed class RecordedOp : IRecordedOp - where TState : struct, IDisposable - { - // not readonly: releasing clears the capture in place, which a defensive copy would not - private TState _state; - private readonly AutoDatabaseAsyncOperation _operation; - private readonly TaskCompletionSource? _proxy; - private Task? _attempt; - - public RecordedOp(in TState state, AutoDatabaseAsyncOperation operation, bool fireAndForget) - { - _state = state; - _operation = operation; - - // Fire-and-forget declines the reply, so there is nothing for a proxy to carry and none is built - - // saving *both* objects, the TaskCompletionSource and the Task it creates in its own constructor, - // on every fire-and-forget operation queued. Proxy hands back the shared completed task instead, - // which is the same instance a plain RedisTransaction returns for this case. - // - // The value is default(TResult) rather than the command's own "no reply" value, which lives inside - // the command implementation and is not reachable from here - a distinction without a difference, - // since neither is an answer from the server. - _proxy = fireAndForget ? null : new(TaskCreationOptions.RunContinuationsAsynchronously); - } - - // the ! is an annotation artifact, not a null: TResult is unconstrained, so Task and - // Task are the same type - Default simply spells its return the other way round - public Task Proxy => _proxy?.Task ?? CompletedTask.Default(null)!; - - public void Replay(IDatabaseAsync inner) => _attempt = _operation(in _state, inner); - - public Task? ForwardSuccess() - { - // fire-and-forget: Forward guards too, so this is here to skip the continuation below rather than - // for correctness - there is no point registering one to settle a proxy that does not exist - if (_proxy is null) return null; - - // ForwardSuccess only runs after Replay, so this cannot be null; `?? throw` says that to the - // compiler as well as the reader, and keeps the failure loud if the order ever changes - var attempt = _attempt ?? throw new InvalidOperationException("Cannot forward a result before the operation has been replayed"); - if (attempt.IsCompleted) - { - Forward(attempt); - return null; - } - - ForwardLater(attempt); - return attempt; - } - - private void ForwardLater(Task attempt) => attempt.ContinueWith( - static (completed, state) => ((RecordedOp)state!).Forward(completed), - this, - CancellationToken.None, - TaskContinuationOptions.ExecuteSynchronously, - TaskScheduler.Default); - - private void Forward(Task attempt) - { - // a null proxy is fire-and-forget; `?.` then skips the argument too, including the Exception - // access that would otherwise observe a fault. Safe only because such an attempt is the shared - // completed task RedisTransaction hands back for F+F, so it cannot fault - see Observe for the - // attempts that can. - if (attempt.IsCanceled) _proxy?.TrySetCanceled(); - else if (attempt.IsFaulted) _proxy?.TrySetException(attempt.Exception!.InnerExceptions); - else _proxy?.TrySetResult(attempt.GetAwaiter().GetResult()); - } - - // observe the (faulted) inner attempt before faulting the durable proxy, so the discarded - // per-attempt task doesn't surface as an unobserved exception - public void Fault(Exception ex) - { - Observe(); - _proxy?.TrySetException(ex); - } - - public void Observe() => _ = _attempt?.Exception; - - public void Release() => _state.Dispose(); - } - - private sealed class RecordedVoidOp : IRecordedOp - where TState : struct, IDisposable - { - // not readonly: releasing clears the capture in place, which a defensive copy would not - private TState _state; - private readonly AutoDatabaseAsyncOperation _operation; - private readonly TaskCompletionSource? _proxy; - private Task? _attempt; - - public RecordedVoidOp(in TState state, AutoDatabaseAsyncOperation operation, bool fireAndForget) - { - _state = state; - _operation = operation; - _proxy = fireAndForget ? null : new(TaskCreationOptions.RunContinuationsAsynchronously); // see RecordedOp - } - - public Task Proxy => _proxy?.Task ?? Task.CompletedTask; - - public void Replay(IDatabaseAsync inner) => _attempt = _operation(in _state, inner); - - public Task? ForwardSuccess() - { - if (_proxy is null) return null; // fire-and-forget: see RecordedOp.ForwardSuccess - - // ForwardSuccess only runs after Replay, so this cannot be null; `?? throw` says that to the - // compiler as well as the reader, and keeps the failure loud if the order ever changes - var attempt = _attempt ?? throw new InvalidOperationException("Cannot forward a result before the operation has been replayed"); - if (attempt.IsCompleted) - { - Forward(attempt); - return null; - } - - ForwardLater(attempt); - return attempt; - } - - private void ForwardLater(Task attempt) => attempt.ContinueWith( - static (completed, state) => ((RecordedVoidOp)state!).Forward(completed), - this, - CancellationToken.None, - TaskContinuationOptions.ExecuteSynchronously, - TaskScheduler.Default); - - private void Forward(Task attempt) - { - // see RecordedOp.Forward for why `?.` is safe here - if (attempt.IsCanceled) _proxy?.TrySetCanceled(); - else if (attempt.IsFaulted) _proxy?.TrySetException(attempt.Exception!.InnerExceptions); - else _proxy?.TrySetResult(true); - } - - // observe the (faulted) inner attempt before faulting the durable proxy, so the discarded - // per-attempt task doesn't surface as an unobserved exception - public void Fault(Exception ex) - { - Observe(); - _proxy?.TrySetException(ex); - } - - public void Observe() => _ = _attempt?.Exception; - - public void Release() => _state.Dispose(); - } - - private sealed class RecordedCondition - { - private readonly Condition _condition; - private ConditionResult? _attempt; - - public RecordedCondition(Condition condition) - { - _condition = condition; - Result = new ConditionResult(condition); - } - - // the durable result handed back to the caller from AddCondition - public ConditionResult Result { get; } - - public void Replay(ITransactionAsync inner) => _attempt = inner.AddCondition(_condition); - - // Unlike the operations, this needs no deferral for a fire-and-forget EXEC: a conditional transaction - // has to know whether its constraints held before it can decide between EXEC and DISCARD, so the - // condition replies are resolved as part of writing it - even when the EXEC itself wants no reply. - public void ForwardSuccess() - { - if (_attempt is not null) Result.SetSatisfied(_attempt.WasSatisfied); - } - } - - // ---- hand-implemented members the generator deliberately skips ---------------------------------- - // (the Wait family, the synchronous IsConnected probe, and the streaming scans). Wait/IsConnected are - // straight pass-throughs; scans cannot participate in a transaction. - // forwarding is not using: these decorators must implement the interface in full, and the - // implementation cannot be dropped while the interface declares it - #pragma warning disable SER308 // Blocking on a task through the library's Wait helpers - void IRedisAsync.Wait(Task task) => _source.Wait(task); - T IRedisAsync.Wait(Task task) => _source.Wait(task); - void IRedisAsync.WaitAll(Task[] tasks) => _source.WaitAll(tasks); - bool IRedisAsync.TryWait(Task task) => _source.TryWait(task); - #pragma warning restore SER308 - - bool IDatabaseAsync.IsConnected(RedisKey key, CommandFlags flags) => _source.IsConnected(key, flags); - - // routing lookup against the underlying database, not a queued transaction operation - Task IDatabaseAsync.IdentifyEndpointAsync(RedisKey key, CommandFlags flags) => _source.IdentifyEndpointAsync(key, flags); - - // nested transactions are not supported (mirrors RedisTransaction) - ITransactionAsync IDatabaseAsync.CreateTransaction(object? asyncState) => throw new NotSupportedException("Nested transactions are not supported"); - - IAsyncEnumerable IDatabaseAsync.HashScanAsync(RedisKey key, RedisValue pattern, int pageSize, long cursor, int pageOffset, CommandFlags flags) => throw new NotSupportedException("Scans cannot be used inside a transaction"); - IAsyncEnumerable IDatabaseAsync.HashScanNoValuesAsync(RedisKey key, RedisValue pattern, int pageSize, long cursor, int pageOffset, CommandFlags flags) => throw new NotSupportedException("Scans cannot be used inside a transaction"); - IAsyncEnumerable IDatabaseAsync.SetScanAsync(RedisKey key, RedisValue pattern, int pageSize, long cursor, int pageOffset, CommandFlags flags) => throw new NotSupportedException("Scans cannot be used inside a transaction"); - IAsyncEnumerable IDatabaseAsync.VectorSetRangeEnumerateAsync(RedisKey key, RedisValue start, RedisValue end, long count, Exclude exclude, CommandFlags flags) => throw new NotSupportedException("Scans cannot be used inside a transaction"); - IAsyncEnumerable IDatabaseAsync.SortedSetScanAsync(RedisKey key, RedisValue pattern, int pageSize, long cursor, int pageOffset, CommandFlags flags) => throw new NotSupportedException("Scans cannot be used inside a transaction"); -} diff --git a/src/StackExchange.Redis/CapturedArgs.cs b/src/StackExchange.Redis/CapturedArgs.cs deleted file mode 100644 index dcdd6fbb3..000000000 --- a/src/StackExchange.Redis/CapturedArgs.cs +++ /dev/null @@ -1,88 +0,0 @@ -using System; -using System.Buffers; -using System.Runtime.CompilerServices; -using System.Runtime.InteropServices; - -namespace StackExchange.Redis; - -/// -/// Helpers for the captured-argument structs emitted by [AutoDatabase(Replays = true)]. -/// -/// -/// A database that replays re-invokes with the same captured arguments after the caller has regained -/// control, and a argument still belongs to that caller - who is entitled to reuse -/// the buffer once the call returns. So a replaying capture takes a shallow copy and gives it back when the -/// operation is finally done with. -/// -/// Shallow: the elements are copied, not anything they refer to. A wrapping a -/// caller's Memory<byte> still points at the caller's bytes, so mutating *those* would still -/// be seen by a replay. That gap closes when serialization moves to the calling thread; this covers the -/// buffer the caller is most likely to reuse. -/// -/// -internal static class CapturedArgs -{ - /// Take a pooled shallow copy of a captured argument. - /// The element type. - /// The caller's memory to copy. - /// A copy owned by us, to be given back via . - public static ReadOnlyMemory Clone(ReadOnlyMemory value) - { - if (value.IsEmpty) return default; // nothing rented, so nothing for Release to give back - - var rented = ArrayPool.Shared.Rent(value.Length); - value.Span.CopyTo(rented); - return new ReadOnlyMemory(rented, 0, value.Length); - } - - /// - public static Memory Clone(Memory value) - { - if (value.IsEmpty) return default; - - var rented = ArrayPool.Shared.Rent(value.Length); - value.Span.CopyTo(rented); - return new Memory(rented, 0, value.Length); - } - - /// Give a cloned argument back to the pool. - /// The element type. - /// The captured field, cleared as it is released. - /// - /// Clears the field first, so a second call has nothing to give back. Returning the same array twice is - /// not an error the pool reports - it simply pools it twice, and two later rents then hand out the same - /// array to different callers - so this is worth the cheek of writing through a readonly field. - /// - /// Sequential only: the clear is not atomic with the read. That is sufficient here because a captured - /// state is a local within a single async flow, never shared. Note also that calling this on a - /// defensive copy - i.e. via a readonly holder - would clear the copy and leak; the generated - /// funnels hold their state in a plain local. - /// - /// - public static void Release(in ReadOnlyMemory field) - { - var captured = field; - Unsafe.AsRef(in field) = default; - Return(captured); - } - - /// - public static void Release(in Memory field) - { - var captured = field; - Unsafe.AsRef(in field) = default; - Return(captured); - } - - private static void Return(ReadOnlyMemory captured) - { - // note the Length check rather than a null check: an empty ReadOnlyMemory reports a non-null, - // zero-length array through TryGetArray, so testing for null here would never be true - if (MemoryMarshal.TryGetArray(captured, out var segment) && segment.Array is { Length: > 0 }) - { - // cleared on return: these elements can hold references (RedisValue, RedisKeyOrValue), and a - // pooled array would otherwise keep the caller's keys and values alive indefinitely - ArrayPool.Shared.Return(segment.Array, clearArray: true); - } - } -} diff --git a/src/StackExchange.Redis/Contexts/RespTransaction.cs b/src/StackExchange.Redis/Contexts/RespTransaction.cs index 6f71f2c8f..834f48487 100644 --- a/src/StackExchange.Redis/Contexts/RespTransaction.cs +++ b/src/StackExchange.Redis/Contexts/RespTransaction.cs @@ -36,7 +36,28 @@ namespace StackExchange.Redis /// exactly this distinction, and §7f, where the handshake collapsed the same way. /// /// - internal sealed class RespTransactionExecutor : RespExecutorBase + /// + /// What needs from the executor underneath it, beyond + /// : the shape is shared by the plain one-connection unit + /// () and the retrying one that replays the unit per attempt + /// (RespRetryTransactionExecutor). + /// + internal interface ITransactionExecutor + { + /// + void AddCondition(Condition condition, Action? onVerdict = null); + + /// + Task ExecuteAsync(CommandFlags flags = CommandFlags.None); + + /// + bool WasWatchConflict { get; } + + /// + CommandFlags AggregateRetryCategory { get; } + } + + internal sealed class RespTransactionExecutor : RespExecutorBase, ITransactionExecutor { private readonly RespExecutorBase _inner; private readonly RespContext _context; @@ -89,7 +110,7 @@ internal RespTransactionExecutor(RespExecutorBase inner, RespContext? context = /// retry: a failed condition means the state is not what you wanted, where a watch conflict means /// it was, and somebody moved first. Only the second is worth re-reading and trying again. /// - internal bool WasWatchConflict => _watchConflict; + public bool WasWatchConflict => _watchConflict; /// Whether this transaction has been executed or discarded; either way, nothing more will be sent. internal bool IsSent @@ -133,7 +154,9 @@ internal override ValueTask SendTypedAsync( var operation = RespPayloadOperation.Rent(handler, this); try { - if (!Enqueue(operation, in request, cancellationToken)) return new ValueTask(default(TResult)!); + // the SHARED completed-default task, not a fresh default(TResult): a declined reply has + // nothing per-call to carry, and TaskBridge hands the backing task out untouched + if (!Enqueue(operation, in request, cancellationToken)) return new ValueTask(CompletedTask.Default(null)!); } catch (Exception ex) { @@ -220,7 +243,7 @@ private bool Enqueue(RespPayloadOperation operation, in RespRequest request, Can /// surface hands the caller a ConditionResult when the condition is added and fills it in /// later, so "which one failed?" stays answerable; a single overall bool cannot say that. /// - internal void AddCondition(Condition condition, Action? onVerdict = null) + public void AddCondition(Condition condition, Action? onVerdict = null) { if (condition is null) throw new ArgumentNullException(nameof(condition)); lock (_sync) @@ -240,7 +263,7 @@ internal void AddCondition(Condition condition, Action? onVerdict = null) /// does it: they live in the conditions rather than the queue, and re-issuing them is what makes a /// replay safe in the first place. /// - internal CommandFlags AggregateRetryCategory + public CommandFlags AggregateRetryCategory { get { @@ -278,7 +301,7 @@ internal CommandFlags AggregateRetryCategory /// Monitor handshakes on result boxes inside the write lock to get the same effect. /// /// - internal async Task ExecuteAsync(CommandFlags flags = CommandFlags.None) + public async Task ExecuteAsync(CommandFlags flags = CommandFlags.None) { List? queue; List? conditions; diff --git a/src/StackExchange.Redis/Database/RedisDatabase.Scans.cs b/src/StackExchange.Redis/Database/RedisDatabase.Scans.cs index 489c9e786..e7080b5df 100644 --- a/src/StackExchange.Redis/Database/RedisDatabase.Scans.cs +++ b/src/StackExchange.Redis/Database/RedisDatabase.Scans.cs @@ -10,10 +10,10 @@ namespace StackExchange.Redis; /// /// /// -/// These are hand-written because [AutoDatabase] skips them, and it skips them for a real -/// reason: deferred execution does not fit capture-and-replay, so an -returning -/// member cannot be generated. That also means the generator cannot see whether they are implemented, so -/// SER352 has never counted them - ScanGapTests is what watches this file instead. +/// These were hand-written from the first day, because [AutoDatabase] skipped them - deferred +/// execution does not fit capture-and-replay, so an -returning member could not +/// be generated. That also meant the generator could not see whether they were implemented, so SER352 never +/// counted them. Generator and rule are both gone now; ScanGapTests still watches this file. /// /// /// Each pair of members returns the same object, because the shipped contract says a scan is both diff --git a/src/StackExchange.Redis/Database/RedisDatabase.Strings.cs b/src/StackExchange.Redis/Database/RedisDatabase.Strings.cs index 502af10c6..51e347de8 100644 --- a/src/StackExchange.Redis/Database/RedisDatabase.Strings.cs +++ b/src/StackExchange.Redis/Database/RedisDatabase.Strings.cs @@ -16,9 +16,10 @@ namespace StackExchange.Redis /// visibly not. /// /// - /// Implementing a member here removes it from the generated set automatically, because - /// [AutoDatabase] skips whatever the class declares itself - so nothing has to be deleted from - /// a list of stubs, and SER352's count falls by itself. + /// During the port, implementing a member here removed it from the generated set automatically, + /// because [AutoDatabase] skipped whatever the class declared itself - so nothing had to be + /// deleted from a list of stubs, and SER352's count fell by itself until both hit zero and the + /// generator was deleted. The per-group layout it drove remains. /// /// /// Note how little each one is. The body is the whole implementation - no message type, no result diff --git a/src/StackExchange.Redis/Database/RedisDatabase.cs b/src/StackExchange.Redis/Database/RedisDatabase.cs index acbe9e2f3..73c75acd1 100644 --- a/src/StackExchange.Redis/Database/RedisDatabase.cs +++ b/src/StackExchange.Redis/Database/RedisDatabase.cs @@ -26,8 +26,9 @@ namespace StackExchange.Redis /// Losing it is an improvement, not a regression. A command added to the interface used to /// produce a generated throw and a warning; now it does not compile until it is implemented, which is /// the same guarantee RedisDatabase has always had and a better one than a tripwire. The - /// generator itself stays - RetryTransaction is a capture-and-replay - /// wrapper, which is what it is really for. + /// generator itself is gone too: its last consumer, the capture-and-replay RetryTransaction, + /// was replaced by RespRetryTransactionExecutor, which replays rendered frames instead of + /// method calls and needs nothing generated. /// /// /// There is no fallback. This once forwarded anything not yet moved to the v3 RedisDatabase, and @@ -184,7 +185,7 @@ CancellationToken IInternalDatabaseAsync.GetNextFailover() /// The operation to wait for. private void Wait(ValueTask pending) => SyncWait.Wait(pending, multiplexer); - // ---- members the generator deliberately skips (see AutoDatabaseGenerator.SkipMethod) ------------- + // ---- members the (now deleted) [AutoDatabase] generator deliberately skipped ------------- /// /// @@ -224,20 +225,13 @@ public ITransaction CreateTransaction(object? asyncState = null) if (this is IBatch) throw new NotSupportedException("Nested transactions are not supported"); // A retrying database replays a transaction AS A UNIT - conditions, queued commands and EXEC - // together - which the per-frame retry executor cannot express; that is RetryTransaction's - // job. It records against a plain view over the executor underneath the retry link, building - // a fresh inner transaction per attempt, and the durable proxy tasks it hands out cannot - // carry an asyncState - so one is refused rather than silently dropped. + // together - which the per-frame retry executor cannot express. RespRetryTransactionExecutor + // records the rendered frames and the conditions, replays them into a fresh one-shot unit per + // attempt, and settles the durable tasks it handed out only when the unit has finished for + // good - which is also why asyncState works here: the tasks the caller holds are ours. if (_inner.Raw.Executor is Availability.RespRetryExecutor retry) { - if ((asyncState ?? AsyncState) is not null) - { - throw new InvalidOperationException( - "Retrying transactions do not support asyncState; the tasks they hand back are not the tasks that were sent to the server."); - } - - var plain = new RedisDatabase(new RespDatabaseContext(_inner.Raw.WithExecutor(retry.Inner)), multiplexer, null); - return new Availability.RetryTransaction(plain, retry.Controller); + return RedisTransaction.CreateRetryingTransaction(_inner, retry, multiplexer, asyncState ?? AsyncState); } // CanWriteTransactions, not CanWriteRuns: an executor can write a batch without being able to diff --git a/src/StackExchange.Redis/Database/RedisTransaction.cs b/src/StackExchange.Redis/Database/RedisTransaction.cs index 4dbf8a06d..14ee0d69a 100644 --- a/src/StackExchange.Redis/Database/RedisTransaction.cs +++ b/src/StackExchange.Redis/Database/RedisTransaction.cs @@ -23,7 +23,7 @@ namespace StackExchange.Redis /// notes §3c and §7r. /// /// - internal sealed class RedisTransaction : RedisBatch, ITransaction, IInternalTransaction + internal sealed class RedisTransaction : RedisBatch, ITransaction { /// /// @@ -32,11 +32,11 @@ internal sealed class RedisTransaction : RedisBatch, ITransaction, IInternalTran /// private protected sealed override DatabaseFeatureFlags OwnFeatures => DatabaseFeatureFlags.Transaction; - private readonly RespTransactionExecutor _executor; + private readonly ITransactionExecutor _executor; private RedisTransaction( RespDatabaseContext queueing, - RespTransactionExecutor executor, + ITransactionExecutor executor, IConnectionMultiplexer multiplexer, object? asyncState) : base(queueing, executor: null, multiplexer, asyncState) @@ -53,6 +53,33 @@ internal static RedisTransaction CreateTransaction(RespDatabaseContext source, I return new RedisTransaction(new RespDatabaseContext(raw.WithExecutor(executor)), executor, multiplexer, asyncState); } + /// Wrap a retrying context so its transaction replays as a unit. + /// + /// + /// The queueing context keeps everything the database's context had EXCEPT the retry link - the + /// unit is what retries, so a per-frame retry inside it would re-send individual queued commands, + /// which is not a retry of anything the caller asked for. The retrying executor shares the + /// database's , so budgets and failover tracking are the + /// same ones a plain command sees. + /// + /// + /// Not reusable, where the plain transaction is: a replayed unit releases its recorded + /// requests when it finishes, so "execute, then queue more onto the same object" has nothing left + /// to compose onto. The capture-replay RetryTransaction made the same call. + /// + /// + internal static RedisTransaction CreateRetryingTransaction( + RespDatabaseContext source, + Availability.RespRetryExecutor retry, + IConnectionMultiplexer multiplexer, + object? asyncState) + { + var raw = source.Raw; + var plain = raw.WithExecutor(retry.Inner); + var executor = new Availability.RespRetryTransactionExecutor(retry.Inner, plain, retry.Controller); + return new RedisTransaction(new RespDatabaseContext(raw.WithExecutor(executor)), executor, multiplexer, asyncState); + } + /// /// /// The result is handed back now and filled in when the check is answered, which is what @@ -72,15 +99,6 @@ public ConditionResult AddCondition(Condition condition) /// public bool WasWatchConflict => _executor.WasWatchConflict; - /// - /// - /// Without this the retry layer cannot retry a transaction at all. It asks the transaction - /// what a replay would do, and a transaction that does not answer is assumed to be - /// - so a transient fault the caller explicitly asked - /// to ride out is rethrown instead, and their retry policy silently does nothing. - /// - CommandFlags IInternalTransaction.GetAggregateRetryCategory() => _executor.AggregateRetryCategory; - /// /// /// Sync-over-async, suppressed for the same reason and with the same reservations as diff --git a/src/StackExchange.Redis/Database/TaskBridge.cs b/src/StackExchange.Redis/Database/TaskBridge.cs index a79cdde96..2d0981b6d 100644 --- a/src/StackExchange.Redis/Database/TaskBridge.cs +++ b/src/StackExchange.Redis/Database/TaskBridge.cs @@ -59,7 +59,16 @@ internal static class TaskBridge /// internal static Task AsTask(this ValueTask pending, object? asyncState, CommandFlags flags) { - if ((flags & CommandFlags.FireAndForget) != 0) asyncState = null; + if ((flags & CommandFlags.FireAndForget) != 0) + { + asyncState = null; + + // a declined reply that is already answered travels exactly as the executor shaped it, + // which is what lets a transaction hand every fire-and-forget caller ONE shared completed + // task instead of a fresh Task.FromResult per queued command; a result-backed ValueTask + // comes out of AsTask as Task.FromResult anyway, so nothing else changes + if (pending.IsCompletedSuccessfully) return pending.AsTask(); + } if (pending.IsCompletedSuccessfully) { diff --git a/src/StackExchange.Redis/Interfaces/IInternalDatabaseAsync.cs b/src/StackExchange.Redis/Interfaces/IInternalDatabaseAsync.cs index 42626b9c0..b6eb0cea3 100644 --- a/src/StackExchange.Redis/Interfaces/IInternalDatabaseAsync.cs +++ b/src/StackExchange.Redis/Interfaces/IInternalDatabaseAsync.cs @@ -30,19 +30,6 @@ internal interface IInternalDatabaseAsync : IDatabaseAsync object? AsyncState { get; } } -/// -/// Exposes transaction-level detail needed by the retry machinery. Implemented by the concrete -/// transaction types so a RetryTransaction can inspect the transaction it is replaying against. -/// -internal interface IInternalTransaction -{ - /// - /// The most side-effecting retry category across all queued operations (excluding WATCH - /// constraints); this describes what replaying the whole transaction would do. - /// - CommandFlags GetAggregateRetryCategory(); -} - internal static class InternalDatabaseExtension { internal static DatabaseFeatureFlags GetFeatures(this IDatabaseAsync database, out string name) diff --git a/tests/StackExchange.Redis.Tests/DatabaseSurfaceTests.cs b/tests/StackExchange.Redis.Tests/DatabaseSurfaceTests.cs index 6b77d1269..511836a8a 100644 --- a/tests/StackExchange.Redis.Tests/DatabaseSurfaceTests.cs +++ b/tests/StackExchange.Redis.Tests/DatabaseSurfaceTests.cs @@ -14,7 +14,7 @@ namespace StackExchange.Redis.Tests; /// at all, if a command were quietly still being forwarded. So these assert the other half: that the /// members those suites call are implemented by the class itself rather than generated. /// -/// [AutoDatabase] emits EXPLICIT interface implementations, so the interface map is what tells the +/// [AutoDatabase] emitted EXPLICIT interface implementations, so the interface map is what told the /// two apart - a hand-written member maps to a public method, a generated one to a private explicit one. /// Nothing else can see the difference, which is the same reason RedisDatabaseTests goes through /// IDatabase rather than the concrete type. @@ -213,18 +213,9 @@ public void NothingIsGeneratedForTheRedisDatabaseAnyMore() Assert.Equal(["CreateTransaction(Object)"], Generated(string.Empty, typeof(IDatabaseAsync))); } - /// - /// The control for every Assert.Empty above: the query really can see a generated member, so - /// "nothing is generated" means nothing is generated rather than "the query stopped working". - /// - /// - /// RetryTransaction is [AutoDatabase(Replays = true)] and is generated - /// on purpose - capture-and-replay wrappers are what the generator is for, and it keeps that - /// job after the transitional database stops needing it. So this control cannot go stale by being - /// right, which is how the previous one kept failing. - /// - [Fact] - public void TheGeneratedQueryCanStillSeeAGeneratedMember() - => Assert.NotEmpty(Generated(typeof(Availability.RetryTransaction), "String", typeof(IDatabaseAsync))); + // The control that used to sit here - "the query really can see a generated member" - went with the + // [AutoDatabase] generator itself: its last consumer, RetryTransaction, was replaced by the frame-level + // RespRetryTransactionExecutor, so NOTHING in the assembly is generated and the Assert.Empty pins above + // now guard against hand-written explicit implementations creeping back rather than generated ones. } diff --git a/tests/StackExchange.Redis.Tests/QueuedResultTests.cs b/tests/StackExchange.Redis.Tests/QueuedResultTests.cs index 69dde2699..bb0fcb5ab 100644 --- a/tests/StackExchange.Redis.Tests/QueuedResultTests.cs +++ b/tests/StackExchange.Redis.Tests/QueuedResultTests.cs @@ -93,16 +93,16 @@ public async Task TransactionQueuedFireAndForgetStillExecutes() /// /// /// - /// This is a regression test for a real divergence. RetryTransaction records every queued call and - /// hands back a durable proxy task; it used to do so without regard to the flags - which it could not see, - /// as they are captured inside the generated per-command state - so a fire-and-forget result stayed - /// incomplete until ExecuteAsync forwarded it. + /// This is a regression test for a real divergence. A retrying transaction records every queued call and + /// hands back a durable proxy task; the capture-replay wrapper this started life against used to do so + /// without regard to the flags - which it could not see, as they were captured inside generated + /// per-command state - so a fire-and-forget result stayed incomplete until ExecuteAsync forwarded it. /// /// /// The effect was that identical caller code returned instantly on a plain transaction and blocked for /// good on a retrying one, which is precisely the drop-in substitution WithRetry is meant to - /// support. The generated state now exposes its CommandFlags (IFlaggedRedisArgs) so the two - /// agree. + /// support. RespRetryTransactionExecutor sees the flags on the rendered request, so the two agree + /// by construction - and this pins that they stay agreeing. /// /// [Fact] diff --git a/tests/StackExchange.Redis.Tests/RespEndToEndTests.cs b/tests/StackExchange.Redis.Tests/RespEndToEndTests.cs index 4c713beff..210b30455 100644 --- a/tests/StackExchange.Redis.Tests/RespEndToEndTests.cs +++ b/tests/StackExchange.Redis.Tests/RespEndToEndTests.cs @@ -567,23 +567,28 @@ public async Task ABatchOfOneStillWorks() Assert.Equal("solo", (string?)await context.Strings.GetAsync(key)); } - /// A retrying TRANSACTION still refuses, and for a reason of its own. + /// A retrying TRANSACTION exposes a working context too, and it queues rather than sends. /// - /// A retrying database replays one operation, so decorating its executor reproduces it exactly. A - /// retrying transaction replays the whole recorded unit - conditions, queued operations and - /// EXEC together - which a per-frame retry executor cannot express: it would re-send individual - /// frames inside a transaction, retrying something nobody asked for. + /// This used to assert a refusal: the capture-and-replay wrapper recorded method calls, so a + /// context - which produces rendered frames - had nothing to record into, and asking for one threw. + /// The frame-replaying executor records rendered frames, which is exactly what a context emits, so the + /// refusal went with the wrapper: a command queued through the context here is recorded, replayed and + /// answered through the same durable proxy as an interface call. /// [Fact] - public async Task ARetryingTransactionStillRefusesAContext() + public async Task ARetryingTransactionContextQueuesInsteadOfSending() { await using var conn = Create(); + var key = Me(); var retrying = conn.GetDatabase().WithRetry( new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }); + await retrying.StringSetAsync(key, "tranned"); var tran = retrying.CreateTransaction(); - var ex = Assert.Throws(() => ((IRespKeyspaceTarget)tran).Context); - Assert.Contains("replayed as a unit", ex.Message); + var pending = ((IRespKeyspaceTarget)tran).Context.Strings.GetAsync(key); + Assert.False(pending.IsCompleted, "DEFERRED-OK: transaction did not send immediately"); + Assert.True(await tran.ExecuteAsync(), "EXEC-OK"); + Assert.Equal("tranned", (string?)await pending); } /// And so does a transaction's, for an ordinary single-frame command. diff --git a/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs b/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs index 0d49de0c8..d8f3b2847 100644 --- a/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs +++ b/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs @@ -481,8 +481,8 @@ public async Task WithRetry_Transaction_PerOpError_FaultsOnlyThatProxy() } // Multi-group + a retryable transaction: A fails every EXEC and is knocked out, so the group reroutes to B. - // Because RetryTransaction builds a *fresh* inner transaction against the currently-active member on each - // attempt, the replay lands on B and commits there - the transaction analogue of the single-command + // Because the retrying executor builds a *fresh* attempt transaction against the currently-active member + // on each replay, the retry lands on B and commits there - the transaction analogue of the single-command // WithRetry_FailsOverBetweenGroupsOnLoading test. [Fact] public async Task WithRetry_Transaction_FailsOverBetweenGroups() diff --git a/tests/StackExchange.Redis.Tests/RetryTests/RetryGuardTests.cs b/tests/StackExchange.Redis.Tests/RetryTests/RetryGuardTests.cs index d13c65e86..113edd653 100644 --- a/tests/StackExchange.Redis.Tests/RetryTests/RetryGuardTests.cs +++ b/tests/StackExchange.Redis.Tests/RetryTests/RetryGuardTests.cs @@ -48,15 +48,15 @@ public async Task WithRetry_CarriesAsyncState() Assert.Same(state, task.AsyncState); await task; - // a transaction created *from* a retrying database still refuses one: it replays as a unit, - // handing out durable proxy tasks that are not the tasks sent to the server, so the state - // cannot be carried - and dropping it silently would be worse than refusing + // ...and a transaction created *from* a retrying database carries it too. The capture-and-replay + // wrapper refused one - its durable proxies could not take a state - but the frame-level replay's + // durable tasks are built by the same AsTask bridge as everything else, so the state travels. var retryDb = conn.GetDatabase().WithRetry(Policy()); - var ex = Assert.Throws(() => retryDb.CreateTransaction(state)); - Log(ex.Message); - - // no state: fine - Assert.NotNull(retryDb.CreateTransaction()); + var tran = retryDb.CreateTransaction(state); + var queued = tran.StringSetAsync("asyncstate:key", "value"); + Assert.Same(state, queued.AsyncState); + Assert.True(await tran.ExecuteAsync()); + Assert.True(await queued); } // Key-prefixing composes with retry: the prefix is applied when the operation is captured, so it diff --git a/tests/StackExchange.Redis.Tests/ScanGapTests.cs b/tests/StackExchange.Redis.Tests/ScanGapTests.cs index 6484e1146..3d9034b1d 100644 --- a/tests/StackExchange.Redis.Tests/ScanGapTests.cs +++ b/tests/StackExchange.Redis.Tests/ScanGapTests.cs @@ -11,14 +11,16 @@ namespace StackExchange.Redis.Tests; /// -/// The deferred-execution members on , which SER352 cannot see. +/// The deferred-execution members on , which SER352 could not see. /// /// /// -/// [AutoDatabase] skips every IEnumerable<T>/IAsyncEnumerable<T> member - -/// deferred execution does not fit capture-and-replay - so these are hand-written, and the generator has -/// no way to tell a real implementation from a forwarding throw. To it they are simply "declared", so the -/// SER352 count omits them entirely. This file is the tripwire in the one place that can see them. +/// [AutoDatabase] skipped every IEnumerable<T>/IAsyncEnumerable<T> member - +/// deferred execution does not fit capture-and-replay - so these were hand-written, and the generator had +/// no way to tell a real implementation from a forwarding throw. To it they were simply "declared", so the +/// SER352 count omitted them entirely. This file was the tripwire in the one place that could see them, +/// and it outlives the generator: the members are still forwarded one at a time, so the gap it watches +/// is still a gap something has to watch. /// /// /// It earned that description immediately: wiring the scans to the context surface made the previous