diff --git a/Directory.Build.props b/Directory.Build.props index 7d2e9c6c5..76557f67f 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -10,7 +10,7 @@ true $(MSBuildThisFileDirectory)Shared.ruleset NETSDK1069 - $(NoWarn);NU5105;NU1507;SER001;SER002;SER003;SER004;SER005;SER006 + $(NoWarn);NU5105;NU1507;SER001;SER002;SER003;SER004;SER005;SER006;SER008 https://stackexchange.github.io/StackExchange.Redis/ReleaseNotes https://stackexchange.github.io/StackExchange.Redis/ MIT diff --git a/docs/HImport.md b/docs/HImport.md new file mode 100644 index 000000000..8455a775b --- /dev/null +++ b/docs/HImport.md @@ -0,0 +1,76 @@ +Hash Import +=== + +The `HIMPORT` command (Redis 8.10 and later) is a fast, session-based way to create many hashes that share a common +set of field names — for example, importing a batch of records where every record has the same columns. Rather than +sending the field names again for every hash (as a series of `HSET` calls would), the field names are declared once per +connection and each hash then supplies only its values, positionally matched to those fields. + +On the wire this is a *connection-sticky* container command with several sub-commands: `HIMPORT PREPARE` registers a +named *fieldset* (the ordered field names) on the current connection, `HIMPORT SET` creates one hash from a row of +values against that fieldset, and `HIMPORT DISCARD` releases the fieldset. The fieldset lives only on the connection +that prepared it and disappears when that connection is reset or closed. + +Because StackExchange.Redis is a multiplexer that owns the connection lifetime on your behalf — you do not control +*which* physical connection a given command travels on† — it does **not** expose the individual `HIMPORT` sub-commands. +Managing a `PREPARE`/`SET`/`DISCARD` sequence yourself would require pinning them all to one connection, which is +exactly the detail the multiplexer abstracts away. Instead, the library exposes a single bulk operation, +`IDatabase.HashImport` (and `HashImportAsync`), that performs the whole import for you: it generates a private fieldset, +issues the `PREPARE`, one `SET` per entry, and the terminating `DISCARD`, all guaranteed to land on a single connection. + +†: *in theory* because of multiplexing there is a single connection, but with automatic reconnects, cluster +slot migrations, active-active / retries, etc: *in reality* it is more complicated than that. + +Usage +--- + +You supply the shared field names once, and then one `HashImportEntry` per hash — each carrying the target key and that +hash's values, in the same order as the field names: + +``` c# +IDatabase db = muxer.GetDatabase(); + +// performance note: you may use leased/oversized buffers throughout, see Notes + +// the field names shared by every hash we are importing +var fields = new RedisValue[] { "name", "email", "age" }; + +// one entry per hash: the key, plus its values positionally matching the fields above +var entries = new HashImportEntry[] +{ + new("user:1", new RedisValue[] { "alice", "a@example.com", 30 }), + new("user:2", new RedisValue[] { "bob", "b@example.com", 25 }), + new("user:3", new RedisValue[] { "carol", "c@example.com", 42 }), +}; + +var failures = await db.HashImportAsync(fields, entries); +foreach (var failure in failures) // empty array on full success +{ + Console.WriteLine($"entry {failure.Index} ({failure.Key}) failed: {failure.Message}"); +} +``` + +After this completes, `user:1`, `user:2` and `user:3` each exist as a hash with the `name`, `email` and `age` fields +set to their respective values. + +Notes +--- + +- `ReadOnlyMemory` is used in the API (rather than arrays) so that you can pass slices of larger buffers without copying — useful + when importing in chunks from a pooled or reused backing array. + - In particular, only 2 leases are needed: a buffer of type `HashImportEntry` of length `entries`, + and a buffer of type `RedisValue` of length `(entries + 1) * fields`, using a [stride](https://en.wikipedia.org/wiki/Stride_of_an_array) of + length `fields` per entry, plus a leading header row of the field names. +- Every entry must supply exactly as many values as there are field names; a mismatch throws (`ArgumentException`) + before anything is sent. +- Each entry **replaces** any existing hash at its key (it does not merge into it) — this is import, not `HSET`. +- The import is **not atomic**. Per-entry failures — for example a key that already holds a non-hash value, which + yields a `WRONGTYPE` error — are returned as `HashImportFailure[]` (each with the entry `Index`, `Key`, and server + `Message`); an empty array means full success, and other entries still apply. A *setup* failure (an invalid field + list) is thrown instead. If you need all-or-nothing semantics, this is not the right tool. +- A zero-entry import is a no-op. There is no single-row shortcut: one entry uses the same `HIMPORT` path so its + replace-semantics are identical to a multi-entry import. +- Inside a `MULTI`/`EXEC` transaction (`CreateTransaction`), the import is unrolled into individual queued commands + (`PREPARE`, one `SET` per entry, then `DISCARD`) so the whole import executes atomically as part of the transaction. + Batches are *not* supported. +- Being connection-sticky, it is not cluster-aware; in a cluster, all supplied keys must map to a single node. diff --git a/docs/exp/SER008.md b/docs/exp/SER008.md new file mode 100644 index 000000000..fec4a2be9 --- /dev/null +++ b/docs/exp/SER008.md @@ -0,0 +1,24 @@ +Redis 8.10 is currently in preview and may be subject to change. + +New features in Redis 8.10: + +- Session-based bulk hash import via fieldsets (`HIMPORT`), surfaced as `IDatabase.HashImport`/`HashImportAsync` + +The corresponding library features must also be considered subject to change: + +1. Existing bindings may cease working correctly if the underlying server API changes. +2. Changes to the server API may require changes to the library API, manifesting in either/both of build-time + or run-time breaks. + +While this seems *unlikely*, it must be considered a possibility. If you acknowledge this, you can suppress +this warning by adding the following to your `csproj` file: + +```xml +$(NoWarn);SER008 +``` + +or more granularly / locally in C#: + +``` c# +#pragma warning disable SER008 +``` diff --git a/docs/index.md b/docs/index.md index 69a4cc70f..2837a0091 100644 --- a/docs/index.md +++ b/docs/index.md @@ -48,6 +48,7 @@ Documentation - [Streams](Streams) - how to use the Stream data type - [Arrays](Arrays) - how to use Redis Arrays as sparse arrays of values - [Vector Sets](VectorSets) - how to use Vector Sets for similarity search with embeddings +- [Hash Import](HImport) - bulk-importing many hashes that share a common set of field names - [Where are `KEYS` / `SCAN` / `FLUSH*`?](KeysScan) - how to use server-based commands - [Profiling](Profiling) - profiling interfaces, as well as how to profile in an `async` world - [Scripting](Scripting) - running Lua scripts with convenient named parameter replacement diff --git a/src/RESPite/Shared/Experiments.cs b/src/RESPite/Shared/Experiments.cs index 304b1f624..0cf540eba 100644 --- a/src/RESPite/Shared/Experiments.cs +++ b/src/RESPite/Shared/Experiments.cs @@ -14,6 +14,7 @@ internal static class Experiments public const string Respite = "SER004"; public const string UnitTesting = "SER005"; public const string Server_8_8 = "SER006"; + public const string Server_8_10 = "SER008"; // ReSharper restore InconsistentNaming diff --git a/src/StackExchange.Redis/APITypes/HashImportEntry.cs b/src/StackExchange.Redis/APITypes/HashImportEntry.cs new file mode 100644 index 000000000..0a9a503b1 --- /dev/null +++ b/src/StackExchange.Redis/APITypes/HashImportEntry.cs @@ -0,0 +1,38 @@ +using System; +using System.Diagnostics.CodeAnalysis; +using RESPite; + +namespace StackExchange.Redis; + +/// +/// Describes a single hash to be created during a bulk operation: a key plus the +/// field values for that key, supplied positionally against the shared field-name list of the import. +/// +/// +/// The are matched positionally to the fields passed to the import, so +/// must equal the number of fields. +/// +[Experimental(Experiments.Server_8_10, UrlFormat = Experiments.UrlFormat)] +public readonly struct HashImportEntry +{ + /// + /// Initializes a value. + /// + /// The key of the hash to create. + /// The field values, in the same order as the field names supplied to the import. + public HashImportEntry(RedisKey key, ReadOnlyMemory values) + { + Key = key; + Values = values; + } + + /// + /// The key of the hash to create. + /// + public RedisKey Key { get; } + + /// + /// The field values for this hash, positionally matched to the shared field names of the import. + /// + public ReadOnlyMemory Values { get; } +} diff --git a/src/StackExchange.Redis/APITypes/HashImportFailure.cs b/src/StackExchange.Redis/APITypes/HashImportFailure.cs new file mode 100644 index 000000000..7cd4136ad --- /dev/null +++ b/src/StackExchange.Redis/APITypes/HashImportFailure.cs @@ -0,0 +1,35 @@ +using System.Diagnostics.CodeAnalysis; +using RESPite; + +namespace StackExchange.Redis; + +/// +/// Describes a single entry that failed during a bulk operation. The import is not +/// atomic, so individual entries can fail (for example, when a target key already holds a non-hash value) while others +/// succeed; returns the failures (an empty array indicates a fully successful import). +/// +[Experimental(Experiments.Server_8_10, UrlFormat = Experiments.UrlFormat)] +public readonly struct HashImportFailure +{ + internal HashImportFailure(int index, RedisKey key, string message) + { + Index = index; + Key = key; + Message = message; + } + + /// + /// The zero-based index of the failing entry within the entries supplied to the import. + /// + public int Index { get; } + + /// + /// The key of the failing entry, in the caller's key-space (any keyspace-isolation prefix has been removed). + /// + public RedisKey Key { get; } + + /// + /// The server error describing why the entry failed (for example, a WRONGTYPE message). + /// + public string Message { get; } +} diff --git a/src/StackExchange.Redis/Enums/RedisCommand.cs b/src/StackExchange.Redis/Enums/RedisCommand.cs index a355c02e6..039671d44 100644 --- a/src/StackExchange.Redis/Enums/RedisCommand.cs +++ b/src/StackExchange.Redis/Enums/RedisCommand.cs @@ -98,6 +98,7 @@ internal enum RedisCommand HGETEX, HGETDEL, HGETALL, + HIMPORT, HINCRBY, HINCRBYFLOAT, HKEYS, @@ -363,6 +364,7 @@ internal static bool IsPrimaryOnly(this RedisCommand command) case RedisCommand.HEXPIREAT: case RedisCommand.HGETDEL: case RedisCommand.HGETEX: + case RedisCommand.HIMPORT: case RedisCommand.HINCRBY: case RedisCommand.HINCRBYFLOAT: case RedisCommand.HMSET: diff --git a/src/StackExchange.Redis/Interfaces/IDatabase.cs b/src/StackExchange.Redis/Interfaces/IDatabase.cs index c0761b9a7..309954024 100644 --- a/src/StackExchange.Redis/Interfaces/IDatabase.cs +++ b/src/StackExchange.Redis/Interfaces/IDatabase.cs @@ -788,6 +788,32 @@ public partial interface IDatabase : IRedis, IDatabaseAsync /// IEnumerable HashScanNoValues(RedisKey key, RedisValue pattern = default, int pageSize = RedisBase.CursorUtils.DefaultLibraryPageSize, long cursor = RedisBase.CursorUtils.Origin, int pageOffset = 0, CommandFlags flags = CommandFlags.None); + /// + /// Bulk-creates many hashes that share a common set of field names, using the server-side session fieldset + /// mechanism (HIMPORT). Each entry supplies a key and the field values for that key, matched positionally + /// against . + /// + /// The field names shared by every hash being imported; every entry must supply exactly this many values. + /// The hashes to create; each carries the key and its field values. + /// The flags to use for this operation. + /// The entries that failed to import (an empty array on full success). + /// + /// + /// The import is not atomic: it is unrolled into a session-local HIMPORT PREPARE, one HIMPORT SET + /// per entry, and a terminating HIMPORT DISCARD, all issued on a single connection. Each SET + /// replaces any existing hash at that key. Per-entry failures (such as a key holding a non-hash value) + /// are returned as values rather than thrown; setup failures (an invalid field + /// list) are thrown. A zero-entry import is a no-op. + /// + /// + /// Inside a MULTI/EXEC transaction the import is unrolled into individual queued commands so it + /// participates atomically; it is not supported inside a batch. Being connection-sticky, it is not cluster-aware. + /// + /// + /// + [Experimental(Experiments.Server_8_10, UrlFormat = Experiments.UrlFormat)] + HashImportFailure[] HashImport(ReadOnlyMemory fields, ReadOnlyMemory entries, CommandFlags flags = CommandFlags.None); + /// /// Sets the specified fields to their respective values in the hash stored at key. /// This command overwrites any specified fields that already exist in the hash, leaving other unspecified fields untouched. diff --git a/src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs b/src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs index 40db55cfd..a67c01a82 100644 --- a/src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs +++ b/src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs @@ -180,6 +180,10 @@ public partial interface IDatabaseAsync : IRedisAsync /// IAsyncEnumerable HashScanNoValuesAsync(RedisKey key, RedisValue pattern = default, int pageSize = RedisBase.CursorUtils.DefaultLibraryPageSize, long cursor = RedisBase.CursorUtils.Origin, int pageOffset = 0, CommandFlags flags = CommandFlags.None); + /// + [Experimental(Experiments.Server_8_10, UrlFormat = Experiments.UrlFormat)] + Task HashImportAsync(ReadOnlyMemory fields, ReadOnlyMemory entries, CommandFlags flags = CommandFlags.None); + /// Task HashSetAsync(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None); diff --git a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs index 952fe0ed3..4267a307a 100644 --- a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs +++ b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs @@ -183,6 +183,9 @@ public Task HashSetAsync(RedisKey key, RedisValue hashField, RedisValue va public Task HashStringLengthAsync(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None) => Inner.HashStringLengthAsync(ToInner(key), hashField, flags); + public async Task HashImportAsync(ReadOnlyMemory fields, ReadOnlyMemory entries, CommandFlags flags = CommandFlags.None) => + ToOuter(await Inner.HashImportAsync(fields, ToInner(entries), flags).ForAwait()); + public Task HashSetAsync(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None) => Inner.HashSetAsync(ToInner(key), hashFields, flags); @@ -859,6 +862,29 @@ public void WaitAll(params Task[] tasks) => protected internal RedisKey ToInner(RedisKey outer) => RedisKey.WithPrefix(Prefix, outer); + // reverse of ToInner: strip the leading key-prefix bytes so a key that came back from the inner database + // (for example, inside a HashImport failure) is reported to the caller in their own, un-prefixed key-space. + protected internal RedisKey ToOuter(RedisKey inner) + { + var prefix = Prefix; + if (prefix is null || prefix.Length == 0) return inner; + byte[]? full = inner; + if (full is null || full.Length < prefix.Length) return inner; // not prefixed as expected; leave as-is + var outer = new byte[full.Length - prefix.Length]; + Buffer.BlockCopy(full, prefix.Length, outer, 0, outer.Length); + return outer; + } + + // maps failure keys back to the caller's key-space, in place (we own the array returned by the inner database) + private protected HashImportFailure[] ToOuter(HashImportFailure[] failures) + { + for (int i = 0; i < failures.Length; i++) + { + failures[i] = new HashImportFailure(failures[i].Index, ToOuter(failures[i].Key), failures[i].Message); + } + return failures; + } + protected RedisKey ToInnerOrDefault(RedisKey outer) => (outer == default(RedisKey)) ? outer : ToInner(outer); @@ -914,6 +940,18 @@ protected RedisKey ToInnerOrDefault(RedisKey outer) => protected KeyValuePair ToInner(KeyValuePair outer) => new KeyValuePair(ToInner(outer.Key), outer.Value); + protected ReadOnlyMemory ToInner(ReadOnlyMemory outer) + { + if (outer.IsEmpty) return outer; + var span = outer.Span; + var inner = new HashImportEntry[span.Length]; + for (int i = 0; i < span.Length; i++) + { + inner[i] = new HashImportEntry(ToInner(span[i].Key), span[i].Values); + } + return inner; + } + [return: NotNullIfNotNull("outer")] protected KeyValuePair[]? ToInner(KeyValuePair[]? outer) { diff --git a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs index 2ca7ca384..a5887b6a4 100644 --- a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs +++ b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs @@ -174,6 +174,9 @@ public bool HashSet(RedisKey key, RedisValue hashField, RedisValue value, When w public long HashStringLength(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None) => Inner.HashStringLength(ToInner(key), hashField, flags); + public HashImportFailure[] HashImport(ReadOnlyMemory fields, ReadOnlyMemory entries, CommandFlags flags = CommandFlags.None) => + ToOuter(Inner.HashImport(fields, ToInner(entries), flags)); + public void HashSet(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None) => Inner.HashSet(ToInner(key), hashFields, flags); diff --git a/src/StackExchange.Redis/Message.cs b/src/StackExchange.Redis/Message.cs index 17ded6c20..8b4470de2 100644 --- a/src/StackExchange.Redis/Message.cs +++ b/src/StackExchange.Redis/Message.cs @@ -7,6 +7,7 @@ using System.Text; using System.Threading; using Microsoft.Extensions.Logging; +using RESPite; using RESPite.Internal; using RESPite.Messages; using StackExchange.Redis.Profiling; @@ -171,6 +172,29 @@ internal void PrepareToResend(ServerEndPoint resendTo, bool isMoved) public RedisCommand Command => command; public virtual string CommandAndKey => Command.ToString(); + [AsciiHash(nameof(SubCommandMetadata))] + internal enum SubCommand + { + [AsciiHash("")] + Unknown = 0, + [AsciiHash("GETNAME")] + GetName, + [AsciiHash("ID")] + Id, + [AsciiHash("INFO")] + Info, + [AsciiHash("SETINFO")] + SetInfo, + [AsciiHash("SETNAME")] + SetName, + } + + protected virtual bool TryGetSubCommand(out SubCommand subCommand) + { + subCommand = SubCommand.Unknown; + return false; + } + /// /// Things with the potential to cause harm, or to reveal configuration information. /// @@ -180,6 +204,21 @@ public bool IsAdmin { switch (Command) { + case RedisCommand.CLIENT when TryGetSubCommand(out var subCommand): + switch (subCommand) + { + case SubCommand.GetName: + case SubCommand.SetName: + case SubCommand.Id: + case SubCommand.Info: + case SubCommand.SetInfo: + return false; + } + return true; + /* possible? reasonable? + case RedisCommand.CONFIG when TryGetSubCommand(out var subCommand): + // allow .Get? + */ case RedisCommand.BGREWRITEAOF: case RedisCommand.BGSAVE: case RedisCommand.CLIENT: @@ -691,7 +730,7 @@ internal bool ComputeResult(PhysicalConnection connection, ref RespReader reader } catch (Exception ex) { - connection.OnDetailLog($"{ex.GetType().Name}: {ex.Message}"); + connection?.OnDetailLog($"{ex.GetType().Name}: {ex.Message}"); ex.Data.Add("got", prefix.ToString()); connection?.BridgeCouldBeNull?.Multiplexer?.OnMessageFaulted(this, ex); box?.SetException(ex); @@ -2017,5 +2056,40 @@ private UnknownMessage() : base(0, CommandFlags.None, RedisCommand.UNKNOWN) { } } public void SetNoFlush() => Flags |= NoFlushFlag; + + internal static partial class SubCommandMetadata + { + [AsciiHash(CaseSensitive = false)] + internal static partial bool TryParse(ReadOnlySpan value, out SubCommand subCommand); + + [AsciiHash(CaseSensitive = false)] + internal static partial bool TryParse(ReadOnlySpan value, out SubCommand subCommand); + + internal static bool TryGetSubCommand(in RedisValue value, out SubCommand subCommand) + { + switch (value.Type) + { + case RedisValue.StorageType.ByteArray: + case RedisValue.StorageType.MemoryManager: + case RedisValue.StorageType.ShortBlob: + // all three contiguous byte-blob kinds expose their bytes directly + // (the discard here *must* be stack-local; that's the "Unsafe" in this API) + return TryParse(value.UnsafeRawSpan(out _), out subCommand); + case RedisValue.StorageType.String: + // char-backed: parse the chars directly, no UTF8 round-trip + return TryParse(value.RawString().AsSpan(), out subCommand); + case RedisValue.StorageType.Sequence when value.GetByteCount() <= BufferBytes: + // non-contiguous: normalize into a small stack buffer + // (sub-commands are short, so anything longer cannot match) + Span tmp = stackalloc byte[BufferBytes]; + var len = value.CopyTo(tmp); + return TryParse(tmp.Slice(0, len), out subCommand); + // numeric / null / unknown are never a sub-command (e.g. it is never `123`); + // if that ever changes, revisit + } + subCommand = SubCommand.Unknown; + return false; + } + } } } diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 7dc5c5811..5bea43583 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -1 +1,14 @@ #nullable enable +[SER008]StackExchange.Redis.HashImportEntry +[SER008]StackExchange.Redis.HashImportEntry.HashImportEntry() -> void +[SER008]StackExchange.Redis.HashImportEntry.HashImportEntry(StackExchange.Redis.RedisKey key, System.ReadOnlyMemory values) -> void +[SER008]StackExchange.Redis.HashImportEntry.Key.get -> StackExchange.Redis.RedisKey +[SER008]StackExchange.Redis.HashImportEntry.Values.get -> System.ReadOnlyMemory +[SER008]StackExchange.Redis.HashImportFailure +[SER008]StackExchange.Redis.HashImportFailure.HashImportFailure() -> void +[SER008]StackExchange.Redis.HashImportFailure.Index.get -> int +[SER008]StackExchange.Redis.HashImportFailure.Key.get -> StackExchange.Redis.RedisKey +[SER008]StackExchange.Redis.HashImportFailure.Message.get -> string! +[SER008]StackExchange.Redis.IDatabase.HashImport(System.ReadOnlyMemory fields, System.ReadOnlyMemory entries, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.HashImportFailure[]! +[SER008]StackExchange.Redis.IDatabaseAsync.HashImportAsync(System.ReadOnlyMemory fields, System.ReadOnlyMemory entries, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +StackExchange.Redis.RedisFeatures.HashImport.get -> bool diff --git a/src/StackExchange.Redis/RedisDatabase.cs b/src/StackExchange.Redis/RedisDatabase.cs index 21518eaa7..f97547800 100644 --- a/src/StackExchange.Redis/RedisDatabase.cs +++ b/src/StackExchange.Redis/RedisDatabase.cs @@ -4,6 +4,7 @@ using System.Diagnostics; using System.Net; using System.Runtime.CompilerServices; +using System.Runtime.InteropServices; using System.Threading.Tasks; using RESPite.Messages; @@ -982,6 +983,97 @@ public Task HashSetAsync(RedisKey key, HashEntry[] hashFields, CommandFlags flag return ExecuteAsync(msg, ResultProcessor.DemandOK); } + public HashImportFailure[] HashImport(ReadOnlyMemory fields, ReadOnlyMemory entries, CommandFlags flags = CommandFlags.None) + { + ValidateHashImport(fields, entries); + if (entries.IsEmpty) return Array.Empty(); + AssertHashImportContext(); + // note: inside a transaction this (correctly) throws via ExecuteSync, as all sync ops do + return ExecuteSync(new HashImportMessage(Database, flags, fields, entries), HashImportProcessor.Default, defaultValue: Array.Empty()); + } + + public Task HashImportAsync(ReadOnlyMemory fields, ReadOnlyMemory entries, CommandFlags flags = CommandFlags.None) + { + ValidateHashImport(fields, entries); + if (entries.IsEmpty) return CompletedTask.FromDefault(Array.Empty(), asyncState); + AssertHashImportContext(); + // a MULTI/EXEC transaction cannot host a nested IMultiMessage, so we unroll into individual queued + // commands instead; a normal connection uses the IMultiMessage form. + if (this is ITransaction) return QueueHashImportInTransaction(fields, entries, flags); + return ExecuteAsync(new HashImportMessage(Database, flags, fields, entries), HashImportProcessor.Default, defaultValue: Array.Empty()); + } + + private void AssertHashImportContext() + { + // transactions are supported (see QueueHashImportInTransaction); batches are not - a batch is a + // fire-and-forget pipeline with no ordering guarantees suitable for the connection-sticky HIMPORT. + if (this is IBatch && this is not ITransaction) + { + throw new NotSupportedException("HashImport is not supported inside a batch; the underlying HIMPORT is connection-sticky and unrolls into multiple ordered commands."); + } + } + + // Inside a MULTI/EXEC we cannot nest the HashImportMessage (an IMultiMessage); instead we enqueue its + // constituent commands - PREPARE, one SET per entry, then DISCARD - as individual transaction operations. + // They execute in order within the single EXEC, on the one transaction connection, so the session-local + // fieldset created by PREPARE is valid for every SET and is cleaned up by DISCARD. Per-entry SET failures are + // gathered into the result (matching the IMultiMessage path); setup failures (PREPARE/DISCARD) are thrown. + private Task QueueHashImportInTransaction(ReadOnlyMemory fields, ReadOnlyMemory entries, CommandFlags flags) + { + var fieldSet = Guid.NewGuid(); + var prepareTask = ExecuteAsync(new HashImportPrepareMessage(Database, flags, fieldSet, fields), ResultProcessor.DemandOK); + var span = entries.Span; + var setTasks = new Task[span.Length]; + var keys = new RedisKey[span.Length]; + for (int i = 0; i < span.Length; i++) + { + keys[i] = span[i].Key; + setTasks[i] = ExecuteAsync(new HashImportSetMessage(Database, flags, fieldSet, span[i].Key, i, span[i].Values), ResultProcessor.DemandOK); + } + var discardTask = ExecuteAsync(new HashImportDiscardMessage(Database, flags, fieldSet), ResultProcessor.Int64); + return AggregateHashImportAsync(prepareTask, setTasks, keys, discardTask); + } + + private static async Task AggregateHashImportAsync(Task prepareTask, Task[] setTasks, RedisKey[] keys, Task discardTask) + { + // observe every task (even on setup failure) to avoid unobserved-exception noise + Exception? setupError = null; + try { await prepareTask.ForAwait(); } + catch (RedisServerException ex) { setupError = ex; } + + List? failures = null; + for (int i = 0; i < setTasks.Length; i++) + { + try { await setTasks[i].ForAwait(); } + catch (RedisServerException ex) when (setupError is null) { (failures ??= new()).Add(new HashImportFailure(i, keys[i], ex.Message)); } + catch (RedisServerException) { /* PREPARE already failed; the cascade of "no such fieldset" errors is noise */ } + } + + try { await discardTask.ForAwait(); } + catch (RedisServerException ex) { setupError ??= ex; } + + if (setupError is not null) throw setupError; + return failures?.ToArray() ?? Array.Empty(); + } + + private static void ValidateHashImport(ReadOnlyMemory fields, ReadOnlyMemory entries) + { + if (entries.IsEmpty) return; // no-op import; nothing to validate + if (fields.IsEmpty) throw new ArgumentException("At least one field name must be supplied.", nameof(fields)); + int fieldCount = fields.Length; + var span = entries.Span; + for (int i = 0; i < span.Length; i++) + { + int valueCount = span[i].Values.Length; + if (valueCount != fieldCount) + { + throw new ArgumentException( + $"Entry {i} supplies {valueCount} value(s) but {fieldCount} field name(s) were provided; the counts must match.", + nameof(entries)); + } + } + } + public Task HashSetIfNotExistsAsync(RedisKey key, RedisValue hashField, RedisValue value, CommandFlags flags) { var msg = Message.Create(Database, flags, RedisCommand.HSETNX, key, hashField, value); @@ -4174,6 +4266,215 @@ private Message GetSortedSetMultiPopMessage(RedisKey[] keys, Order order, long c } } + // HIMPORT is connection-sticky: a session-local fieldset (identified by a per-import GUID) is PREPAREd, one + // SET is issued per entry against that fieldset, and the fieldset is finally DISCARDed - all on one connection. + // This is modelled as an IMultiMessage: the sub-messages capture per-entry SET errors (into the failures list) + // and setup errors (PREPARE) into the parent; the terminal DISCARD (this) carries the user-facing result - + // returning the collected failures, or throwing when setup failed. + internal sealed class HashImportMessage : Message, IMultiMessage + { + private readonly ReadOnlyMemory _fields; + private readonly ReadOnlyMemory _entries; + private readonly Guid _fieldSet; + private string? _setupError; + private List? _failures; + + public HashImportMessage(int db, CommandFlags flags, ReadOnlyMemory fields, ReadOnlyMemory entries) + : base(db, flags, RedisCommand.HIMPORT) + { + _fields = fields; + _entries = entries; + _fieldSet = Guid.NewGuid(); // unique per import; carried as a Guid and written as 16 raw bytes, no allocation + SetNoRedirect(); // every sub-command must stay on this one connection + } + + // note: replies are processed in order on the single read loop, so no synchronization is needed here + internal void RecordSetFailure(int index, in RedisKey key, string? error) => (_failures ??= new()).Add(new HashImportFailure(index, key, error ?? string.Empty)); + internal void RecordSetupError(string? error) => _setupError ??= error; + internal string? SetupError => _setupError; + internal HashImportFailure[] BuildResult() => _failures?.ToArray() ?? Array.Empty(); + + public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy) + { + int slot = ServerSelectionStrategy.NoSlot; + var span = _entries.Span; + for (int i = 0; i < span.Length; i++) + { + slot = serverSelectionStrategy.CombineSlot(slot, span[i].Key); + } + return slot; + } + + public IEnumerable GetMessages(PhysicalConnection connection) + { + var step = new HashImportStepProcessor(this); + + // HIMPORT PREPARE
+ var prepare = new HashImportPrepareMessage(Db, Flags, _fieldSet, _fields); + prepare.SetNoRedirect(); + prepare.SetSource(step, null); + yield return prepare; + + // HIMPORT SET
+ for (int i = 0; i < _entries.Length; i++) + { + var entry = _entries.Span[i]; + var set = new HashImportSetMessage(Db, Flags, _fieldSet, entry.Key, i, entry.Values); + set.SetNoRedirect(); + set.SetSource(step, null); + yield return set; + } + + // HIMPORT DISCARD
(this) - cleans up the session fieldset and carries the final result + yield return this; + } + + protected override void WriteImpl(in MessageWriter writer) + { + WriteDiscard(writer, _fieldSet); + } + + public override int ArgCount => 2; + } + + // standalone HIMPORT DISCARD, used when the import is unrolled into a MULTI/EXEC transaction (where the + // IMultiMessage root cannot be used); see QueueHashImportInTransaction. + internal sealed class HashImportDiscardMessage : Message + { + private readonly Guid _fieldSet; + + public HashImportDiscardMessage(int db, CommandFlags flags, Guid fieldSet) + : base(db, flags, RedisCommand.HIMPORT) + { + _fieldSet = fieldSet; + } + + protected override void WriteImpl(in MessageWriter writer) => WriteDiscard(writer, _fieldSet); + + public override int ArgCount => 2; + } + + private static void WriteDiscard(in MessageWriter writer, Guid fieldSet) + { + writer.WriteHeader(RedisCommand.HIMPORT, 2); + writer.WriteBulkString(RedisLiterals.DISCARD); + WriteFieldSet(writer, fieldSet); + } + + // writes a fieldset GUID as a 16-byte bulk string with no allocation, uniformly across all target frameworks + // (the token is opaque - the server treats it as an arbitrary byte string - so the exact byte layout is + // irrelevant, only that PREPARE / SET / DISCARD within one import serialize the same GUID identically). + private static void WriteFieldSet(in MessageWriter writer, Guid fieldSet) + { + Span buffer = stackalloc byte[16]; + Unsafe.WriteUnaligned(ref buffer[0], fieldSet); + writer.WriteBulkString(buffer); + } + + internal sealed class HashImportPrepareMessage : Message + { + private readonly Guid _fieldSet; + private readonly ReadOnlyMemory _fields; + + public HashImportPrepareMessage(int db, CommandFlags flags, Guid fieldSet, ReadOnlyMemory fields) + : base(db, flags, RedisCommand.HIMPORT) + { + _fieldSet = fieldSet; + _fields = fields; + } + + protected override void WriteImpl(in MessageWriter writer) + { + var fields = _fields.Span; + writer.WriteHeader(RedisCommand.HIMPORT, 2 + fields.Length); + writer.WriteBulkString(RedisLiterals.PREPARE); + WriteFieldSet(writer, _fieldSet); + for (int i = 0; i < fields.Length; i++) writer.WriteBulkString(fields[i]); + } + + public override int ArgCount => 2 + _fields.Length; + } + + internal sealed class HashImportSetMessage : Message.CommandKeyBase + { + private readonly Guid _fieldSet; + private readonly ReadOnlyMemory _values; + + public HashImportSetMessage(int db, CommandFlags flags, Guid fieldSet, in RedisKey key, int index, ReadOnlyMemory values) + : base(db, flags, RedisCommand.HIMPORT, key) + { + _fieldSet = fieldSet; + Index = index; + _values = values; + } + + // carried so a failing SET can be reported back with its originating entry index and key + internal int Index { get; } + internal RedisKey EntryKey => Key; + + protected override void WriteImpl(in MessageWriter writer) + { + var values = _values.Span; + writer.WriteHeader(RedisCommand.HIMPORT, 3 + values.Length); + writer.WriteBulkString(RedisLiterals.SET); + writer.Write(Key); + WriteFieldSet(writer, _fieldSet); + for (int i = 0; i < values.Length; i++) writer.WriteBulkString(values[i]); + } + + public override int ArgCount => 3 + _values.Length; + } + + // processor for the PREPARE / SET sub-messages: success (+OK) is discarded; a server error on a SET is recorded + // as a per-entry failure (with its index + key), while a PREPARE error is recorded as a setup error - both on + // the parent, for the terminal DISCARD to surface. + internal sealed class HashImportStepProcessor : ResultProcessor + { + private readonly HashImportMessage _parent; + public HashImportStepProcessor(HashImportMessage parent) => _parent = parent; + + public override bool SetResult(PhysicalConnection connection, Message message, ref RespReader reader) + { + reader.MovePastBof(); + if (reader.IsError) + { + var error = reader.ReadString(); + if (message is HashImportSetMessage set) + { + _parent.RecordSetFailure(set.Index, set.EntryKey, error); + } + else + { + _parent.RecordSetupError(error); + } + } + message.SetResponseReceived(); + return true; // always consumed; never fault the connection over a per-step error + } + + protected override bool SetResultCore(PhysicalConnection connection, Message message, ref RespReader reader) => true; + } + + // processor for the terminal HIMPORT DISCARD: the integer reply is irrelevant; if a setup step (PREPARE) failed + // the whole operation throws, otherwise the collected per-entry failures are returned as the result. + internal sealed class HashImportProcessor : ResultProcessor + { + public static readonly HashImportProcessor Default = new(); + private HashImportProcessor() { } + + protected override bool SetResultCore(PhysicalConnection connection, Message message, ref RespReader reader) + { + if (message is not HashImportMessage himport) return false; + if (himport.SetupError is { } error) + { + SetException(message, new RedisServerException(error)); + return true; + } + SetResult(message, himport.BuildResult()); + return true; + } + } + private ITransaction? GetLockExtendTransaction(RedisKey key, RedisValue value, TimeSpan expiry) { var tran = CreateTransactionIfAvailable(asyncState); @@ -5613,6 +5914,24 @@ public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy) return slot; } public override int ArgCount => _args.Count; + + protected override bool TryGetSubCommand(out SubCommand subCommand) + { + // the sub-command (if any) is the first argument after the command itself, + // e.g. CLIENT [GETNAME]; ad-hoc Execute args are boxed objects, so normalize + // the first one to a RedisValue before probing it against the known sub-commands + foreach (object arg in _args) + { + var value = RedisValue.TryParse(arg, out var valid); + if (valid) + { + return SubCommandMetadata.TryGetSubCommand(value, out subCommand); + } + break; // only the first argument is a sub-command candidate + } + subCommand = SubCommand.Unknown; + return false; + } } private sealed class ScriptEvalMessage : Message, IMultiMessage diff --git a/src/StackExchange.Redis/RedisFeatures.cs b/src/StackExchange.Redis/RedisFeatures.cs index 1a40cd427..b5233d843 100644 --- a/src/StackExchange.Redis/RedisFeatures.cs +++ b/src/StackExchange.Redis/RedisFeatures.cs @@ -50,7 +50,8 @@ namespace StackExchange.Redis v8_2_0_rc1 = new Version(8, 1, 240), // 8.2 RC1 is version 8.1.240 v8_4_0_rc1 = new Version(8, 3, 224), // 8.4 RC1 is version 8.3.224 v8_6_0 = new Version(8, 6, 0), - v8_8_0 = new Version(8, 7, 225); // 8.8 is version 8.7.225 + v8_8_0 = new Version(8, 7, 225), // 8.8 is version 8.7.225 + v8_10_0 = new Version(8, 9, 241); // 8.10 preview is version 8.9.241 #pragma warning restore SA1310 // Field names should not contain underscore #pragma warning restore SA1311 // Static readonly fields should begin with upper-case letter @@ -298,6 +299,11 @@ public RedisFeatures(Version version) ///
public bool DeleteWithValueCheck => Version.IsAtLeast(v8_4_0_rc1); + /// + /// Is session-based bulk hash import (HIMPORT) available? + /// + public bool HashImport => Version.IsAtLeast(v8_10_0); + #pragma warning restore 1629 // Documentation text should end with a period. /// diff --git a/src/StackExchange.Redis/RedisLiterals.cs b/src/StackExchange.Redis/RedisLiterals.cs index 9709eb02e..bbe3a3294 100644 --- a/src/StackExchange.Redis/RedisLiterals.cs +++ b/src/StackExchange.Redis/RedisLiterals.cs @@ -32,6 +32,7 @@ public static readonly RedisValue DESC = RedisValue.FromRaw("DESC"u8), DIFF = RedisValue.FromRaw("DIFF"u8), DIFF1 = RedisValue.FromRaw("DIFF1"u8), + DISCARD = RedisValue.FromRaw("DISCARD"u8), DOCTOR = RedisValue.FromRaw("DOCTOR"u8), ENCODING = RedisValue.FromRaw("ENCODING"u8), EX = RedisValue.FromRaw("EX"u8), @@ -93,6 +94,7 @@ public static readonly RedisValue PAUSE = RedisValue.FromRaw("PAUSE"u8), PERSIST = RedisValue.FromRaw("PERSIST"u8), PING = RedisValue.FromRaw("PING"u8), + PREPARE = RedisValue.FromRaw("PREPARE"u8), PURGE = RedisValue.FromRaw("PURGE"u8), PX = RedisValue.FromRaw("PX"u8), PXAT = RedisValue.FromRaw("PXAT"u8), diff --git a/tests/StackExchange.Redis.Tests/HashImportTests.cs b/tests/StackExchange.Redis.Tests/HashImportTests.cs new file mode 100644 index 000000000..03bbf42ba --- /dev/null +++ b/tests/StackExchange.Redis.Tests/HashImportTests.cs @@ -0,0 +1,285 @@ +using System; +using System.Threading.Tasks; +using StackExchange.Redis.KeyspaceIsolation; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Integration tests for / +/// (the session-based HIMPORT bulk-import feature, Redis 8.10+). +/// +[RunPerProtocol] +public class HashImportTests(ITestOutputHelper output, SharedConnectionFixture fixture) : TestBase(output, fixture) +{ + private static readonly RedisValue[] Fields = ["name", "email", "age"]; + + [Fact] + public async Task ImportsManyHashes() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + var prefix = Me(); + + RedisKey k1 = prefix + ":1", k2 = prefix + ":2", k3 = prefix + ":3"; + await db.KeyDeleteAsync([k1, k2, k3]); + + HashImportEntry[] entries = + [ + new(k1, new RedisValue[] { "alice", "a@example.com", 30 }), + new(k2, new RedisValue[] { "bob", "b@example.com", 25 }), + new(k3, new RedisValue[] { "carol", "c@example.com", 42 }), + ]; + + var result = await db.HashImportAsync(Fields, entries); + Assert.Empty(result); + + Assert.Equal("alice", await db.HashGetAsync(k1, "name")); + Assert.Equal("a@example.com", await db.HashGetAsync(k1, "email")); + Assert.Equal(30, (int)await db.HashGetAsync(k1, "age")); + Assert.Equal("bob", await db.HashGetAsync(k2, "name")); + Assert.Equal("carol", await db.HashGetAsync(k3, "name")); + Assert.Equal(3, await db.HashLengthAsync(k3)); + } + + [Fact] + public async Task SingleEntry_Works() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + RedisKey key = Me(); + await db.KeyDeleteAsync(key); + + HashImportEntry[] entries = [new(key, new RedisValue[] { "alice", "a@example.com", 30 })]; + var result = await db.HashImportAsync(Fields, entries); + Assert.Empty(result); + + Assert.Equal("alice", await db.HashGetAsync(key, "name")); + Assert.Equal(3, await db.HashLengthAsync(key)); + } + + [Fact] + public async Task SingleEntry_ReplacesExistingHash() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + RedisKey key = Me(); + await db.KeyDeleteAsync(key); + await db.HashSetAsync(key, [new("old", 1), new("keep", 2)]); + + // single-entry now uses HIMPORT (not HSET), so it replaces rather than merges - consistent with multi-entry + await db.HashImportAsync(Fields, new HashImportEntry[] { new(key, new RedisValue[] { "alice", "a@x", 30 }) }); + + Assert.Equal(3, await db.HashLengthAsync(key)); + Assert.False(await db.HashExistsAsync(key, "old")); + } + + [Fact] + public async Task ZeroEntries_IsNoOp() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + // no server round-trip expected; simply completes + await db.HashImportAsync(Fields, Array.Empty()); + await db.HashImportAsync(default, default); // fully empty is also a no-op + } + + [Fact] + public async Task KeyPrefixIsolation_PrefixesEntryKeys() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var prefix = Me() + ":"; + var db = conn.GetDatabase().WithKeyPrefix(prefix); + var raw = conn.GetDatabase(); + + const string inner = "u1"; + string full = prefix + inner; + await raw.KeyDeleteAsync(full); + + await db.HashImportAsync(Fields, new HashImportEntry[] { new(inner, new RedisValue[] { "alice", "a@x", 30 }) }); + + // written under the prefixed key + Assert.Equal("alice", await raw.HashGetAsync(full, "name")); + } + + [Fact] + public async Task KeyPrefixIsolation_FailureKeyIsUnprefixed() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var prefix = Me() + ":"; + var db = conn.GetDatabase().WithKeyPrefix(prefix); + var raw = conn.GetDatabase(); + + const string inner = "u1"; + string full = prefix + inner; + await raw.KeyDeleteAsync(full); + await raw.StringSetAsync(full, "not-a-hash"); // wrong type under the prefixed key -> WRONGTYPE + + var failures = await db.HashImportAsync(Fields, new HashImportEntry[] { new(inner, new RedisValue[] { "a", "b", "c" }) }); + + var failure = Assert.Single(failures); + Assert.Equal(0, failure.Index); + // reported in the caller's (un-prefixed) key-space, i.e. "u1" - NOT the prefixed key sent to the server + Assert.Equal(inner, (string?)failure.Key); + Assert.StartsWith("WRONGTYPE", failure.Message); + } + + [Fact] + public void MismatchedValueCount_ThrowsBeforeSending() + { + using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + // 3 fields but only 1 value in the entry -> synchronous ArgumentException + HashImportEntry[] entries = [new(Me(), new RedisValue[] { "only-one" })]; + // validation runs synchronously, before any Task is returned + Assert.Throws(() => { _ = db.HashImportAsync(Fields, entries); }); + } + + [Fact] + public async Task DuplicateFieldNames_SurfacesServerError() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + RedisValue[] dupFields = ["f1", "f1"]; + HashImportEntry[] entries = + [ + new(Me() + ":1", new RedisValue[] { "a", "b" }), + new(Me() + ":2", new RedisValue[] { "c", "d" }), + ]; + await Assert.ThrowsAsync(() => db.HashImportAsync(dupFields, entries)); + } + + [Fact] + public async Task WorksInsideTransaction() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + var prefix = Me(); + RedisKey k1 = prefix + ":1", k2 = prefix + ":2"; + await db.KeyDeleteAsync([k1, k2]); + + var tran = db.CreateTransaction(); + var importTask = tran.HashImportAsync(Fields, new HashImportEntry[] + { + new(k1, new RedisValue[] { "alice", "a@x", 30 }), + new(k2, new RedisValue[] { "bob", "b@x", 25 }), + }); + // the hashes must not exist until the transaction executes + Assert.False(await db.KeyExistsAsync(k1)); + + Assert.True(await tran.ExecuteAsync()); + Assert.Empty(await importTask); + + Assert.Equal("alice", await db.HashGetAsync(k1, "name")); + Assert.Equal("bob", await db.HashGetAsync(k2, "name")); + Assert.Equal(3, await db.HashLengthAsync(k2)); + } + + [Fact] + public async Task SingleEntryWorksInsideTransaction() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + RedisKey key = Me(); + await db.KeyDeleteAsync(key); + + var tran = db.CreateTransaction(); + var importTask = tran.HashImportAsync(Fields, new HashImportEntry[] { new(key, new RedisValue[] { "alice", "a@x", 30 }) }); + Assert.True(await tran.ExecuteAsync()); + Assert.Empty(await importTask); + + Assert.Equal("alice", await db.HashGetAsync(key, "name")); + Assert.Equal(3, await db.HashLengthAsync(key)); + } + + [Fact] + public async Task ExistingHashIsReplacedNotMerged() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + var prefix = Me(); + RedisKey k1 = prefix + ":1", k2 = prefix + ":2"; + await db.KeyDeleteAsync([k1, k2]); + // pre-existing hash with extra fields that are NOT part of the import + await db.HashSetAsync(k1, [new("old", 1), new("keep", 2)]); + + await db.HashImportAsync(Fields, new HashImportEntry[] + { + new(k1, new RedisValue[] { "alice", "a@x", 30 }), + new(k2, new RedisValue[] { "bob", "b@x", 25 }), + }); + + // HIMPORT SET replaces the whole hash: the pre-existing 'old'/'keep' fields are gone + Assert.Equal(3, await db.HashLengthAsync(k1)); + Assert.False(await db.HashExistsAsync(k1, "old")); + Assert.Equal("alice", await db.HashGetAsync(k1, "name")); + } + + [Fact] + public async Task WrongTypeKey_ReportedAsPerEntryFailure() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + var prefix = Me(); + RedisKey k0 = prefix + ":0", k1 = prefix + ":1"; + await db.KeyDeleteAsync([k0, k1]); + await db.StringSetAsync(k0, "i-am-a-string"); // wrong type for a hash import + + // per-entry failures are returned, not thrown + var result = await db.HashImportAsync(Fields, new HashImportEntry[] + { + new(k0, new RedisValue[] { "alice", "a@x", 30 }), // index 0: WRONGTYPE + new(k1, new RedisValue[] { "bob", "b@x", 25 }), // index 1: fine + }); + + var failure = Assert.Single(result); + Assert.Equal(0, failure.Index); + Assert.Equal(k0, failure.Key); + Assert.StartsWith("WRONGTYPE", failure.Message); + + // the string key is untouched... + Assert.Equal("i-am-a-string", await db.StringGetAsync(k0)); + // ...and the other (valid) entry was still written (the import is not atomic) + Assert.Equal("bob", await db.HashGetAsync(k1, "name")); + } + + [Fact] + public async Task WrongTypeKey_ReportedAsFailure_InsideTransaction() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + var prefix = Me(); + RedisKey k0 = prefix + ":0", k1 = prefix + ":1"; + await db.KeyDeleteAsync([k0, k1]); + await db.StringSetAsync(k0, "i-am-a-string"); + + var tran = db.CreateTransaction(); + var importTask = tran.HashImportAsync(Fields, new HashImportEntry[] + { + new(k0, new RedisValue[] { "alice", "a@x", 30 }), + new(k1, new RedisValue[] { "bob", "b@x", 25 }), + }); + Assert.True(await tran.ExecuteAsync()); + var result = await importTask; + + var failure = Assert.Single(result); + Assert.Equal(0, failure.Index); + Assert.Equal(k0, failure.Key); + Assert.StartsWith("WRONGTYPE", failure.Message); + Assert.Equal("bob", await db.HashGetAsync(k1, "name")); + } + + [Fact] + public async Task NotSupportedInsideBatch() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var batch = conn.GetDatabase().CreateBatch(); + HashImportEntry[] entries = + [ + new(Me() + ":1", new RedisValue[] { "a", "b", "c" }), + new(Me() + ":2", new RedisValue[] { "d", "e", "f" }), + ]; + // the batch/transaction guard runs synchronously + Assert.Throws(() => { _ = batch.HashImportAsync(Fields, entries); }); + } +} diff --git a/tests/StackExchange.Redis.Tests/LibraryNameSuffixAdminTests.cs b/tests/StackExchange.Redis.Tests/LibraryNameSuffixAdminTests.cs new file mode 100644 index 000000000..eafb99acb --- /dev/null +++ b/tests/StackExchange.Redis.Tests/LibraryNameSuffixAdminTests.cs @@ -0,0 +1,54 @@ +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +public class LibraryNameSuffixAdminTests(InProcServerFixture fixture) +{ + private ConfigurationOptions NoAdminConfig() + { + // start from the shared in-process server config, but explicitly *disable* admin mode + var options = fixture.Config.Clone(); + options.AllowAdmin = false; + Assert.False(options.AllowAdmin); + return options; + } + + [Fact] + public async Task AddLibraryNameSuffixWorksWithoutAdmin() + { + await using var conn = await ConnectionMultiplexer.ConnectAsync(NoAdminConfig()); + + // internally this fixes up connected servers via CLIENT SETINFO (best-effort); the + // CLIENT SETINFO sub-command is not admin; don't report it (telemetry, etc) + conn.AddLibraryNameSuffix("mysuffix"); + } + + [Fact] + public async Task ClientSubCommandsViaExecuteDoNotRequireAdmin() + { + await using var conn = await ConnectionMultiplexer.ConnectAsync(NoAdminConfig()); + var server = conn.GetServer(conn.GetEndPoints()[0]); + + // none of these CLIENT sub-commands are admin, so they must not trip the admin-mode guard + // even though AllowAdmin is disabled (regression: the ad-hoc ExecuteMessage previously did + // not expose its sub-command to Message.IsAdmin, so CLIENT was treated as wholesale-admin) + var id = server.Execute("CLIENT", "ID"); + Assert.True((long)id > 0); + + Assert.Equal("OK", (string?)server.Execute("CLIENT", "SETNAME", "roundtrip")); + Assert.Equal("roundtrip", (string?)server.Execute("CLIENT", "GETNAME")); + } + + [Fact] + public async Task AdminClientSubCommandsStillRequireAdmin() + { + await using var conn = await ConnectionMultiplexer.ConnectAsync(NoAdminConfig()); + var server = conn.GetServer(conn.GetEndPoints()[0]); + + // CLIENT LIST is a genuine admin sub-command (not in the allow-list), so it must still be + // blocked when AllowAdmin is disabled - the fix must not blanket-allow every CLIENT usage + var ex = Assert.Throws(() => server.Execute("CLIENT", "LIST")); + Assert.Contains("admin mode", ex.Message); + } +} diff --git a/tests/StackExchange.Redis.Tests/ResultProcessorUnitTests/HashImport.cs b/tests/StackExchange.Redis.Tests/ResultProcessorUnitTests/HashImport.cs new file mode 100644 index 000000000..2d308ccbd --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ResultProcessorUnitTests/HashImport.cs @@ -0,0 +1,84 @@ +using Xunit; + +namespace StackExchange.Redis.Tests.ResultProcessorUnitTests; + +/// +/// Tests for the internal processors that back : the per-step processor that +/// captures a failing SET (as a per-entry failure) or PREPARE (as a setup error) into the parent, and the terminal +/// processor that surfaces those (returning the failures array, or throwing on setup error). +/// +public class HashImport(ITestOutputHelper log) : ResultProcessorUnitTest(log) +{ + private static RedisDatabase.HashImportMessage NewParent() + => new(0, CommandFlags.None, default, default); + + private static RedisDatabase.HashImportSetMessage NewSet(int index, RedisKey key) + => new(0, CommandFlags.None, default, key, index, default); + + [Fact] + public void Step_Ok_RecordsNothing() + { + var parent = NewParent(); + var step = new RedisDatabase.HashImportStepProcessor(parent); + Execute("+OK\r\n", step, message: NewSet(0, "k")); + Assert.Null(parent.SetupError); + Assert.Empty(parent.BuildResult()); + } + + [Fact] + public void Step_SetError_IsCapturedAsFailureWithIndexAndKey() + { + var parent = NewParent(); + var step = new RedisDatabase.HashImportStepProcessor(parent); + Execute("-WRONGTYPE Operation against a key holding the wrong kind of value\r\n", step, message: NewSet(3, "user:3")); + + var failure = Assert.Single(parent.BuildResult()); + Assert.Equal(3, failure.Index); + Assert.Equal("user:3", failure.Key); + Assert.StartsWith("WRONGTYPE", failure.Message); + Assert.Null(parent.SetupError); // a SET failure is not a setup error + } + + [Fact] + public void Step_PrepareError_IsCapturedAsSetupError() + { + var parent = NewParent(); + var step = new RedisDatabase.HashImportStepProcessor(parent); + // a non-SET message (e.g. PREPARE) failing is a setup error + Execute("-ERR duplicate field name in fieldset\r\n", step, message: DummyMessage()); + Assert.Equal("ERR duplicate field name in fieldset", parent.SetupError); + Assert.Empty(parent.BuildResult()); + } + + [Fact] + public void Terminal_Success_NoFailures() + { + var parent = NewParent(); + var result = Execute(":1\r\n", RedisDatabase.HashImportProcessor.Default, message: parent); + Assert.Empty(result!); + } + + [Fact] + public void Terminal_ReturnsCollectedFailures() + { + var parent = NewParent(); + parent.RecordSetFailure(1, "k1", "WRONGTYPE nope"); + parent.RecordSetFailure(4, "k4", "WRONGTYPE also nope"); + + var result = Execute(":1\r\n", RedisDatabase.HashImportProcessor.Default, message: parent)!; + Assert.Equal(2, result.Length); + Assert.Equal(1, result[0].Index); + Assert.Equal(4, result[1].Index); + } + + [Fact] + public void Terminal_SetupError_Throws() + { + var parent = NewParent(); + parent.RecordSetupError("ERR duplicate field name in fieldset"); + + Assert.False(TryExecute(":1\r\n", RedisDatabase.HashImportProcessor.Default, out _, out var ex, message: parent)); + var server = Assert.IsType(ex); + Assert.Equal("ERR duplicate field name in fieldset", server.Message); + } +} diff --git a/tests/StackExchange.Redis.Tests/RoundTripUnitTests/HashImport.cs b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/HashImport.cs new file mode 100644 index 000000000..7c4be6fb2 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/HashImport.cs @@ -0,0 +1,50 @@ +using System; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests.RoundTripUnitTests; + +/// +/// Verifies the exact wire bytes of the individual HIMPORT sub-messages that +/// unrolls into. The fieldset GUID is written as a 16-byte bulk string; +/// gives a deterministic run of 16 zero bytes to assert against. +/// +public class HashImport(ITestOutputHelper log) +{ + private static readonly string FieldSet = "$16\r\n" + new string('\0', 16) + "\r\n"; + + [Fact(Timeout = 5000)] + public async Task Prepare_RoundTrips() + { + ReadOnlyMemory fields = new RedisValue[] { "name", "email", "age" }; + var msg = new RedisDatabase.HashImportPrepareMessage(0, CommandFlags.None, Guid.Empty, fields); + + // HIMPORT PREPARE
name email age + var request = "*6\r\n$7\r\nHIMPORT\r\n$7\r\nPREPARE\r\n" + FieldSet + "$4\r\nname\r\n$5\r\nemail\r\n$3\r\nage\r\n"; + var result = await TestConnection.ExecuteAsync(msg, ResultProcessor.DemandOK, request, "+OK\r\n", log: log); + Assert.True(result); + } + + [Fact(Timeout = 5000)] + public async Task Set_RoundTrips() + { + ReadOnlyMemory values = new RedisValue[] { "v1", "v2" }; + var msg = new RedisDatabase.HashImportSetMessage(0, CommandFlags.None, Guid.Empty, (RedisKey)"user:1", 0, values); + + // HIMPORT SET user:1
v1 v2 (key is arg index 2 per the server key-spec) + var request = "*6\r\n$7\r\nHIMPORT\r\n$3\r\nSET\r\n$6\r\nuser:1\r\n" + FieldSet + "$2\r\nv1\r\n$2\r\nv2\r\n"; + var result = await TestConnection.ExecuteAsync(msg, ResultProcessor.DemandOK, request, "+OK\r\n", log: log); + Assert.True(result); + } + + [Fact(Timeout = 5000)] + public async Task Set_SingleValue_RoundTrips() + { + ReadOnlyMemory values = new RedisValue[] { "only" }; + var msg = new RedisDatabase.HashImportSetMessage(0, CommandFlags.None, Guid.Empty, (RedisKey)"k", 0, values); + + var request = "*5\r\n$7\r\nHIMPORT\r\n$3\r\nSET\r\n$1\r\nk\r\n" + FieldSet + "$4\r\nonly\r\n"; + var result = await TestConnection.ExecuteAsync(msg, ResultProcessor.DemandOK, request, "+OK\r\n", log: log); + Assert.True(result); + } +}