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..af61d5712 --- /dev/null +++ b/docs/HImport.md @@ -0,0 +1,97 @@ +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-local* container command: `HIMPORT PREPARE` registers a named *field-set* (the +ordered field names) on the current connection, `HIMPORT SET` creates one hash from a row of values against that +field-set, and `HIMPORT DISCARD` releases it. A field-set lives only on the connection that prepared it and disappears +when that connection is reset or closed. + +StackExchange.Redis is a multiplexer: it owns the connection lifetime on your behalf, and you do not control *which* +physical connection a given command travels on†. Rather than hide this behind an all-in-one bulk call, the library +exposes something close to the raw shape — a reusable [`HashImport`](xref:StackExchange.Redis.HashImport) field-set plus +a per-row `IDatabase.HashImport` — and takes care of the one hard part for you: the `HIMPORT PREPARE` is **injected +automatically** on whichever connection a given import actually writes to (exactly the way a `SELECT` is injected for +database selection). A transparent reconnect, or a fan-out to another cluster node, simply re-prepares on demand. You +never manage `PREPARE`/`SET`/`DISCARD` ordering or connection pinning yourself. + +†: *in theory* multiplexing means a single connection, but with automatic reconnects, cluster slot +migrations, active-active / retries, etc: *in reality* it is more complicated than that — which is exactly why the raw +connection-local commands are not safe to drive directly through a multiplexer. + +Usage +--- + +Create a field-set once (declaring the shared field names, in order), then import each hash by supplying its key and +its values positionally against those fields: + +``` c# +IDatabase db = muxer.GetDatabase(); + +// declare the field names shared by every hash we are importing; reusable and safe to share/keep +await using var fields = HashImport.Create("name", "email", "age"); + +// import as many hashes as you like - streamed, not materialized up front, so the total is unbounded +await db.HashImportAsync("user:1", fields, new RedisValue[] { "alice", "a@example.com", 30 }); +await db.HashImportAsync("user:2", fields, new RedisValue[] { "bob", "b@example.com", 25 }); +await db.HashImportAsync("user:3", fields, new RedisValue[] { "carol", "c@example.com", 42 }); +``` + +After these complete, `user:1`, `user:2` and `user:3` each exist as a hash with the `name`, `email` and `age` fields set +to their respective values. Disposing the field-set (`await using` / `Dispose`) sends a best-effort `HIMPORT DISCARD` to +release the server-side state; it is optional (the state also dies with the connection) but good hygiene for long-lived +connections. + +### Reusing a values buffer + +To avoid allocating a fresh array per row, you may reuse a single `RedisValue[]` (or a slice of a larger pooled buffer), +refilling it for each hash. If you do, you **must await each import before refilling the buffer for the next +row**‡ — the library reads the values when the command is actually written to the socket, which (on the +async path) can be *after* `HashImportAsync` returns, so awaiting the returned task is what guarantees the library has +finished with your data and the buffer is safe to overwrite: + +``` c# +await using var fields = HashImport.Create("name", "email", "age"); +var row = new RedisValue[3]; // reused for every hash + +foreach (var record in records) +{ + row[0] = record.Name; + row[1] = record.Email; + row[2] = record.Age; + await db.HashImportAsync(record.Key, fields, row); // MUST await before the next refill +} +``` + +Do **not** fire off many `HashImportAsync` calls sharing one buffer without awaiting between them (nor pass the buffer +fire-and-forget) — overwriting it while a prior write is still pending corrupts the data on the wire. + +‡: we hope to lift this restriction in a future release, so that a shared buffer can be refilled without +waiting for each round-trip. + +Notes +--- + +- The field-set is created without touching the server; the `HIMPORT PREPARE` is injected lazily on first use per + connection. A `HashImport` is immutable and may be used concurrently and against multiple databases/multiplexers + (useful for active-active). The field-set name on the wire is an opaque, process-unique id, so it never collides with + another field-set even on a shared connection. +- Field names are validated at `Create`: **duplicate** names are rejected (the server rejects them too, but only via the + injected — fire-and-forget — `PREPARE`, so we fail fast instead), and **null** names are rejected. An **empty** field + name is allowed (a hash may legitimately have an empty-string field). +- `ReadOnlyMemory` is used for the values (rather than an array) so you can pass slices of a larger, pooled + or reused buffer without copying — useful when importing in chunks. +- Each call must supply exactly as many values as the field-set has fields; a mismatch throws (`ArgumentException`) + before anything is sent. +- Each import **replaces** any existing hash at its key (it does not merge into it) — this is import, not `HSET`. +- Each import is applied **on its own** and may be pipelined freely with unrelated work. A server error (for example a + key already holding a non-hash value, giving `WRONGTYPE`) is thrown for that call as usual, and does not affect other + imports; unless the call is fire-and-forget, in which case you have opted out of seeing the result. +- **Cluster-aware**: each key routes to its slot as normal, re-preparing the field-set per node as needed — you do *not* + need hash tags to keep keys together. +- Supported inside a **batch** (an ordered pipeline). *Not* supported inside a `MULTI`/`EXEC` **transaction**: the + connection-local `PREPARE` cannot be staged inside the transaction, so it throws `NotSupportedException`. diff --git a/docs/HashTags.md b/docs/HashTags.md new file mode 100644 index 000000000..0ad8ef453 --- /dev/null +++ b/docs/HashTags.md @@ -0,0 +1,57 @@ +Hash Tags and Slots +=== + +Outside of Redis Cluster (a single server, or a standalone primary/replica) there are no "slots", so none of this applies - +this topic is specific to Redis Cluster environments. + +In a Redis Cluster the keyspace is divided into **16384 hash slots**; every key belongs to exactly one slot, and every +slot is owned by exactly one node at any moment. The slot for a key is based on the CRC hash of the key. Clients (including +StackExchange.Redis) use this to route each command to the node that owns the key's slot; if a key has moved, the server +replies `MOVED`/`ASK` and the client re-routes. + +Because a slot lives wholly on one node, **co-locating keys in the same slot guarantees they are on the same node** - which +is what multi-key operations need. + +Multi-key operations +--- + +Operations that involve multiple keys (`MSET`, `SUNIONSTORE`, etc - or `MULTI`/`EXEC` batches) typically require *all* the keys +they touch to be in the same slot, because they cannot span nodes. An attempted multi-slot operation is rejected with a `CROSSSLOT` error. + +Hash tags +--- + +A **hash tag** lets you force otherwise-different keys into the *same* slot. If a key contains a (non-empty) `{...}` section, only the +bytes **between the first `{` and the next `}`** are hashed, instead of the whole key: + +| Key | Bytes hashed | +| --- | --- | +| `user:1:profile` | `user:1:profile` (the whole key) | +| `{user:1}:profile` | `user:1` | +| `{user:1}:settings` | `user:1` | +| `{user:1}:orders` | `user:1` | + +So `{user:1}:profile`, `{user:1}:settings` and `{user:1}:orders` all hash to the same slot and can participate in one +transaction, script, or bulk import. Note that as a consequence: *all the data with the same hash tag must now fit +on that one shard*; you can't "cheat" the slot rules by using `{dummy}real_key_here`, because that simply pushes your +entire database onto a single node, defeating the entire point of Redis Cluster. Hash-tags are best used to co-locate +genuinely related data - for example splitting things by customer, tenant, etc. + +Inspecting the slot +--- + +`IConnectionMultiplexer.HashSlot(RedisKey)` returns the slot a key resolves to, which is handy for verifying that keys +you intend to group really do share a slot: + +``` c# +var slot1 = muxer.HashSlot("{user:1}:profile"); +var slot2 = muxer.HashSlot("{user:1}:orders"); +// slot1 == slot2 +``` + +Interaction with key prefixes +--- + +A [key prefix](KeysValues) (via `WithKeyPrefix`) is prepended to the key *before* hashing, so it participates in the +slot calculation like any other bytes. To co-locate prefixed keys, make sure the hash tag falls inside the portion that +ends up shared — for example prefix `"{tenant42}:"` puts every key under that prefix in one slot. 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..a21164785 100644 --- a/docs/index.md +++ b/docs/index.md @@ -48,6 +48,8 @@ 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 +- [Hash Tags and Slots](HashTags) - co-locating keys in the same cluster slot for multi-key operations - [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/HashImport.cs b/src/StackExchange.Redis/APITypes/HashImport.cs new file mode 100644 index 000000000..2e896a3aa --- /dev/null +++ b/src/StackExchange.Redis/APITypes/HashImport.cs @@ -0,0 +1,266 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using RESPite; + +namespace StackExchange.Redis; + +/// +/// A reusable, connection-local field-set for streaming hash imports via . +/// +/// +/// +/// The underlying HIMPORT family is session-local: a field-set is PREPAREd against a single +/// physical connection and then referenced by name from each HIMPORT SET. Because the multiplexer hides +/// connections (pooling, transparent reconnects, cluster routing, active/active), this type does not pin a +/// connection. Instead, the PREPARE is injected lazily and automatically on whichever connection a given +/// actually writes to (mirroring how SELECT is injected for database +/// selection); a reconnect or a fan-out to another cluster node simply re-prepares on demand. Each import is applied +/// on its own and may be pipelined freely with unrelated work, so imports are effectively unbounded. +/// +/// +/// Disposing sends a best-effort HIMPORT DISCARD to the connections the field-set was prepared on, releasing the +/// server-side state. Disposal is not required for correctness — the state also dies with the connection — but is good +/// hygiene for long-lived connections. +/// +/// A single is safe to use concurrently and against multiple databases/multiplexers. +/// +[Experimental(Experiments.Server_8_10, UrlFormat = Experiments.UrlFormat)] +public sealed class HashImport : IDisposable, IAsyncDisposable +{ + // process-wide monotonic id; deliberately not bound to any multiplexer so a single field-set can span + // active/active deployments. The id doubles as the opaque, connection-local field-set name on the wire - its 8 raw + // bytes are written verbatim (binary-safe) whenever a name is needed, so no separate byte[] is stored. + private static long _counter; + + private readonly ReadOnlyMemory _fields; + private readonly long _id; + + private readonly object _sync = new(); + // servers this field-set has been prepared against, weakly held so an idle multiplexer can still be collected; + // used only to target DISCARD on disposal. Guarded by _sync. (A named struct rather than a ValueTuple: the shipped + // assembly must not reference System.ValueTuple - it breaks .NET Framework; see SanityChecks.ValueTupleNotReferenced.) + private List? _servers; + private volatile bool _disposed; + + private readonly struct ServerRef(WeakReference server, int db) + { + public WeakReference Server { get; } = server; + public int Db { get; } = db; + } + + private HashImport(ReadOnlyMemory fields) + { + if (fields.IsEmpty) throw new ArgumentException("At least one field name must be supplied.", nameof(fields)); + + // Snapshot into storage we own. The field-set is long-lived and its wire encoding must stay stable for the + // object's lifetime, so we must not alias caller memory that could be mutated after Create - which would also + // silently bypass the validation below. A field-set is created once and reused for many rows, so this one + // copy is amortized to nothing (and a handful of RedisValue at that). + var snapshot = fields.ToArray(); + + // The server rejects a PREPARE with duplicate field names, but because PREPARE is injected fire-and-forget that + // would surface only indirectly as every SET failing with "no such fieldset". Reject it up front for a clear + // error at the point of the mistake. + var seen = new HashSet(); + for (int i = 0; i < snapshot.Length; i++) + { + if (snapshot[i].IsNull) throw new ArgumentException("Field names must not be null.", nameof(fields)); + if (!seen.Add(snapshot[i])) throw new ArgumentException($"Duplicate field name: '{snapshot[i]}'.", nameof(fields)); + } + + _fields = snapshot; + _id = Interlocked.Increment(ref _counter); + } + + /// + /// Creates a field-set describing the ordered field names shared by every hash imported through it. + /// + /// The field names; import values are supplied positionally in this order. + public static HashImport Create(params RedisValue[] fields) => new(fields); + + /// + public static HashImport Create(ReadOnlyMemory fields) => new(fields); + + internal long Id => _id; + internal ReadOnlyMemory Fields => _fields; + internal int FieldCount => _fields.Length; + + // writes the opaque field-set name: the id's 8 raw bytes as a bulk string. Endianness is irrelevant (the server + // treats the name as an arbitrary byte string, and a token never leaves the process), so an unaligned blit of the + // id is enough - and identical for this token's every PREPARE/SET/DISCARD, which is all that matters. + internal void WriteName(in MessageWriter writer) + { + Span name = stackalloc byte[8]; + Unsafe.WriteUnaligned(ref name[0], _id); + writer.WriteBulkString(name); + } + + // rejects use of a disposed field-set before anything is sent; a disposed field-set may already have been DISCARDed + // on the server, so a SET against it would silently mis-behave (and would never be cleaned up). + internal void ThrowIfDisposed() + { + if (_disposed) throw new ObjectDisposedException(nameof(HashImport)); + } + + // creates the HIMPORT PREPARE injected (fire-and-forget) ahead of a SET on a connection that has not yet + // prepared this field-set; its result is never surfaced - a genuinely broken PREPARE re-appears as the SET + // failing with a "no such field-set" server error. + internal Message CreatePrepareMessage(int db) => new HashImportPrepareMessage(db, CommandFlags.FireAndForget, this); + + // Records (once per server) that this field-set is now prepared somewhere on the given server, so disposal can + // target a DISCARD there. + // + // Note the deliberate split of responsibilities, in case a future reader worries that ServerEndPoint (which + // outlives any one connection and spans reconnects/nodes) is too coarse to track connection-local state: + // * Correctness - the "must I inject a PREPARE?" decision - is keyed on the actual PhysicalConnection + // (PhysicalConnection._preparedFieldSets). That is the real thing tied to the real session; a reconnect gives a + // fresh, empty set and re-prepares automatically. This list is NEVER consulted for that. + // * This list is used ONLY for best-effort cleanup on Dispose, at node granularity. DISCARD is idempotent and + // carries this field-set's globally-unique id, so it can only ever drop its own state or no-op ("no such + // fieldset") - it can never disturb another field-set regardless of which connection it lands on. If the + // connection rotated since PREPARE, the old session state is already gone and the DISCARD is a harmless no-op + // on the new connection. Node granularity is also exactly right for cluster, where one field-set is prepared + // independently on several nodes; deduping by server keeps this list bounded to the nodes touched (rather than + // growing per reconnect) and naturally follows each node's current connection. + // + // Bookkeeping only - it is called from inside the bridge write lock (during PREPARE injection), so it must never + // itself issue I/O. If the token is already being disposed we simply skip: a field-set prepared by a SET racing + // against Dispose may be stranded, but it dies with the connection anyway (best-effort), and using a token + // concurrently with disposing it is a caller error. + internal void RegisterServer(ServerEndPoint server, int db) + { + lock (_sync) + { + if (_disposed) return; + _servers ??= new(); + for (int i = 0; i < _servers.Count; i++) + { + if (_servers[i].Server.TryGetTarget(out var existing) && ReferenceEquals(existing, server)) return; + } + _servers.Add(new ServerRef(new WeakReference(server), db)); + } + } + + private List? TakeServers() + { + lock (_sync) + { + if (_disposed) return null; + _disposed = true; + var servers = _servers; + _servers = null; + return servers; + } + } + + /// + /// Releases the connection-local server state for this field-set (best-effort HIMPORT DISCARD). + /// + public void Dispose() + { + var servers = TakeServers(); + if (servers is null) return; + foreach (var entry in servers) + { + if (entry.Server.TryGetTarget(out var server)) _ = SafeDiscardAsync(server, entry.Db); + } + } + + /// + public async ValueTask DisposeAsync() + { + var servers = TakeServers(); + if (servers is null) return; + foreach (var entry in servers) + { + if (entry.Server.TryGetTarget(out var server)) await SafeDiscardAsync(server, entry.Db).ForAwait(); + } + } + + private async Task SafeDiscardAsync(ServerEndPoint server, int db) + { + try + { + await server.WriteDirectAsync(new HashImportDiscardMessage(db, CommandFlags.FireAndForget, this), ResultProcessor.DemandOK).ForAwait(); + } + catch + { + // best-effort: the field-set dies with the connection regardless, so cleanup failures are benign + } + } +} + +// HIMPORT SET : the user-facing per-row import. Carries a reference to its field-set so +// the write path can inject a PREPARE the first time this field-set is seen on a connection (see PhysicalBridge). +internal sealed class HashImportSetMessage : Message.CommandKeyBase +{ + private readonly HashImport _fieldSet; + private readonly ReadOnlyMemory _values; + + public HashImportSetMessage(int db, CommandFlags flags, HashImport fieldSet, in RedisKey key, ReadOnlyMemory values) + : base(db, flags, RedisCommand.HIMPORT, key) + { + _fieldSet = fieldSet; + _values = values; + } + + internal HashImport FieldSet => _fieldSet; + + 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); + _fieldSet.WriteName(writer); + for (int i = 0; i < values.Length; i++) writer.WriteBulkString(values[i]); + } + + public override int ArgCount => 3 + _values.Length; +} + +// HIMPORT PREPARE : injected fire-and-forget ahead of the first SET for a field-set on a +// connection; defines the connection-local name->fields mapping the SET references. +internal sealed class HashImportPrepareMessage : Message +{ + private readonly HashImport _fieldSet; + + public HashImportPrepareMessage(int db, CommandFlags flags, HashImport fieldSet) + : base(db, flags, RedisCommand.HIMPORT) => _fieldSet = fieldSet; + + protected override void WriteImpl(in MessageWriter writer) + { + var fields = _fieldSet.Fields.Span; + writer.WriteHeader(RedisCommand.HIMPORT, 2 + fields.Length); + writer.WriteBulkString(RedisLiterals.PREPARE); + _fieldSet.WriteName(writer); + for (int i = 0; i < fields.Length; i++) writer.WriteBulkString(fields[i]); + } + + public override int ArgCount => 2 + _fieldSet.FieldCount; +} + +// HIMPORT DISCARD : targeted cleanup of a single field-set, issued on disposal. Deliberately not +// DISCARDALL, which would drop sibling field-sets sharing the connection. +internal sealed class HashImportDiscardMessage : Message +{ + private readonly HashImport _fieldSet; + + public HashImportDiscardMessage(int db, CommandFlags flags, HashImport fieldSet) + : base(db, flags, RedisCommand.HIMPORT) => _fieldSet = fieldSet; + + internal long FieldSetId => _fieldSet.Id; + + protected override void WriteImpl(in MessageWriter writer) + { + writer.WriteHeader(RedisCommand.HIMPORT, 2); + writer.WriteBulkString(RedisLiterals.DISCARD); + _fieldSet.WriteName(writer); + } + + public override int ArgCount => 2; +} 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..6f73eeae8 100644 --- a/src/StackExchange.Redis/Interfaces/IDatabase.cs +++ b/src/StackExchange.Redis/Interfaces/IDatabase.cs @@ -788,6 +788,27 @@ 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); + /// + /// Creates a single hash from values supplied positionally against a reusable field-set, + /// using the server-side session field-set mechanism (HIMPORT SET). The HIMPORT PREPARE for + /// is injected automatically the first time it is seen on a connection. + /// + /// The key of the hash to create; any existing hash at this key is replaced. + /// The field-set describing the ordered field names shared by imported hashes. + /// The field values for this hash, matched positionally against the field-set (same count). + /// The flags to use for this operation. + /// + /// + /// Each call is applied on its own and may be freely pipelined with unrelated work, so imports are effectively + /// unbounded. A server error (for example, a key already holding a non-hash value) is thrown as usual unless the + /// call is fire-and-forget. Cluster-aware: each key routes to its slot, re-preparing per node as needed. + /// + /// Not supported inside a transaction (the connection-local PREPARE cannot be staged in MULTI/EXEC). + /// + /// + [Experimental(Experiments.Server_8_10, UrlFormat = Experiments.UrlFormat)] + void HashImport(RedisKey key, HashImport fieldSet, ReadOnlyMemory values, 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..cb3b2a436 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(RedisKey key, HashImport fieldSet, ReadOnlyMemory values, 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..c50c8a852 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 Task HashImportAsync(RedisKey key, HashImport fieldSet, ReadOnlyMemory values, CommandFlags flags = CommandFlags.None) => + Inner.HashImportAsync(ToInner(key), fieldSet, values, flags); + public Task HashSetAsync(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None) => Inner.HashSetAsync(ToInner(key), hashFields, flags); diff --git a/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs b/src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs index 2ca7ca384..2dcfbebc0 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 void HashImport(RedisKey key, HashImport fieldSet, ReadOnlyMemory values, CommandFlags flags = CommandFlags.None) => + Inner.HashImport(ToInner(key), fieldSet, values, flags); + public void HashSet(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None) => Inner.HashSet(ToInner(key), hashFields, flags); diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 21047efa2..f1906d1a8 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -1463,6 +1463,33 @@ private void SelectDatabaseInsideWriteLock(PhysicalConnection connection, Messag } } + // HIMPORT is connection-local: a field-set must be PREPAREd on the same physical connection before a SET can + // reference it. We inject that PREPARE lazily here (mirroring SELECT injection): the first SET for a given + // field-set on a given connection triggers a fire-and-forget PREPARE ahead of it; a reconnect starts with a + // fresh (empty) connection set and re-prepares transparently. A DISCARD drops the id so the connection's set + // stays bounded to live field-sets. All of this runs inside the write lock, so ordering PREPARE-before-SET on + // the wire is guaranteed and the connection's set needs no further synchronization. + private void PrepareFieldSetInsideWriteLock(PhysicalConnection connection, Message message) + { + if (message is HashImportSetMessage set) + { + var fieldSet = set.FieldSet; + if (connection.TryAddPreparedFieldSet(fieldSet.Id)) + { + var prepare = fieldSet.CreatePrepareMessage(message.Db); + connection.EnqueueInsideWriteLock(prepare); + prepare.WriteTo(connection); + prepare.SetRequestSent(); + IncrementOpCount(); + fieldSet.RegisterServer(ServerEndPoint, message.Db); + } + } + else if (message is HashImportDiscardMessage discard) + { + connection.RemovePreparedFieldSet(discard.FieldSetId); + } + } + private WriteResult WriteMessageToServerInsideWriteLock(PhysicalConnection connection, Message message) { if (message == null) @@ -1549,6 +1576,7 @@ private WriteResult WriteMessageToServerInsideWriteLock(PhysicalConnection conne { Debug.Assert(!message.IsHighIntegrity, "prior high integrity message found during transaction?"); } + if (cmd is RedisCommand.HIMPORT) PrepareFieldSetInsideWriteLock(connection, message); connection.EnqueueInsideWriteLock(message); isQueued = true; message.WriteTo(connection); diff --git a/src/StackExchange.Redis/PhysicalConnection.cs b/src/StackExchange.Redis/PhysicalConnection.cs index e03932c4f..bb5c93ee2 100644 --- a/src/StackExchange.Redis/PhysicalConnection.cs +++ b/src/StackExchange.Redis/PhysicalConnection.cs @@ -51,6 +51,11 @@ private static readonly Message private volatile int currentDatabase = 0; + // ids of HIMPORT field-sets already PREPAREd on this specific connection; lazily allocated, only ever touched + // inside the bridge write lock. A fresh connection (e.g. after a reconnect) starts empty, so the write path + // transparently re-prepares on demand - the connection-local server state is scoped to exactly this socket. + private HashSet? _preparedFieldSets; + private ReadMode currentReadMode = ReadMode.NotSpecified; private int failureReported; @@ -703,6 +708,14 @@ internal static Message GetSelectDatabaseCommand(int targetDatabase) : Message.Create(targetDatabase, CommandFlags.FireAndForget, RedisCommand.SELECT); } + // HIMPORT field-set tracking; only ever called inside the bridge write lock, so no synchronization is needed. + // Returns true when the id was not already prepared on this connection (i.e. the caller must inject a PREPARE). + internal bool TryAddPreparedFieldSet(long id) => (_preparedFieldSets ??= new()).Add(id); + + // drops a field-set id when its DISCARD is written, keeping the set bounded to live field-sets over the life of + // a long-lived connection. + internal void RemovePreparedFieldSet(long id) => _preparedFieldSets?.Remove(id); + internal int GetSentAwaitingResponseCount() { lock (_writtenAwaitingResponse) diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 7dc5c5811..508e82da9 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -1 +1,9 @@ #nullable enable +[SER008]StackExchange.Redis.HashImport +[SER008]StackExchange.Redis.HashImport.Dispose() -> void +[SER008]StackExchange.Redis.HashImport.DisposeAsync() -> System.Threading.Tasks.ValueTask +[SER008]StackExchange.Redis.IDatabase.HashImport(StackExchange.Redis.RedisKey key, StackExchange.Redis.HashImport! fieldSet, System.ReadOnlyMemory values, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> void +[SER008]StackExchange.Redis.IDatabaseAsync.HashImportAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.HashImport! fieldSet, System.ReadOnlyMemory values, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task! +[SER008]static StackExchange.Redis.HashImport.Create(params StackExchange.Redis.RedisValue[]! fields) -> StackExchange.Redis.HashImport! +[SER008]static StackExchange.Redis.HashImport.Create(System.ReadOnlyMemory fields) -> StackExchange.Redis.HashImport! +StackExchange.Redis.RedisFeatures.HashImport.get -> bool diff --git a/src/StackExchange.Redis/RedisDatabase.cs b/src/StackExchange.Redis/RedisDatabase.cs index 96390ac1e..7b265266d 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,36 @@ public Task HashSetAsync(RedisKey key, HashEntry[] hashFields, CommandFlags flag return ExecuteAsync(msg, ResultProcessor.DemandOK); } + public void HashImport(RedisKey key, HashImport fieldSet, ReadOnlyMemory values, CommandFlags flags = CommandFlags.None) + => ExecuteSync(GetHashImportMessage(key, fieldSet, values, flags), ResultProcessor.DemandOK); + + public Task HashImportAsync(RedisKey key, HashImport fieldSet, ReadOnlyMemory values, CommandFlags flags = CommandFlags.None) + => ExecuteAsync(GetHashImportMessage(key, fieldSet, values, flags), ResultProcessor.DemandOK); + + private HashImportSetMessage GetHashImportMessage(in RedisKey key, HashImport fieldSet, ReadOnlyMemory values, CommandFlags flags) + { + if (fieldSet is null) throw new ArgumentNullException(nameof(fieldSet)); + fieldSet.ThrowIfDisposed(); + if (values.Length != fieldSet.FieldCount) + { + throw new ArgumentException( + $"{values.Length} value(s) were supplied but the field-set defines {fieldSet.FieldCount} field(s); the counts must match.", + nameof(values)); + } + // HIMPORT needs its PREPARE and SET on the same physical connection; the multiplexer provides that via + // write-path injection (exactly like SELECT). That injection cannot be reused inside a MULTI: every queued + // command adds a slot to the positional EXEC result array, but the transaction only accounts for the ops it + // knows about - an injected PREPARE would desync every following result. (This is the same reason + // SELECT-injection is itself forbidden inside a transaction.) Supporting it would require enqueuing PREPARE + // as a tracked transaction op, which is deliberately out of scope. Batches are fine: an ordered pipeline + // with no EXEC aggregate, so PREPARE injects per connection as normal. + if (this is ITransaction) + { + throw new NotSupportedException("HashImport is not supported inside a transaction; the connection-local HIMPORT PREPARE cannot be injected into a MULTI/EXEC without desyncing the EXEC result array."); + } + return new HashImportSetMessage(Database, flags, fieldSet, key, values); + } + public Task HashSetIfNotExistsAsync(RedisKey key, RedisValue hashField, RedisValue value, CommandFlags flags) { var msg = Message.Create(Database, flags, RedisCommand.HSETNX, key, hashField, value); 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..19ecb20e6 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/HashImportTests.cs @@ -0,0 +1,229 @@ +using System; +using System.Threading.Tasks; +using StackExchange.Redis.KeyspaceIsolation; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Integration tests for / and the +/// reusable field-set (the session-based HIMPORT feature, Redis 8.10+). +/// +[RunPerProtocol] +public class HashImportTests(ITestOutputHelper output, SharedConnectionFixture fixture) : TestBase(output, fixture) +{ + private static RedisValue[] Values(string name, string email, int age) => [name, email, age]; + + [Fact] + public async Task ImportsManyHashesReusingOneFieldSet() + { + 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]); + + await using var fieldSet = HashImport.Create("name", "email", "age"); + await db.HashImportAsync(k1, fieldSet, Values("alice", "a@example.com", 30)); + await db.HashImportAsync(k2, fieldSet, Values("bob", "b@example.com", 25)); + await db.HashImportAsync(k3, fieldSet, Values("carol", "c@example.com", 42)); + + 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); + + await using var fieldSet = HashImport.Create("name", "email", "age"); + await db.HashImportAsync(key, fieldSet, Values("alice", "a@example.com", 30)); + + 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(); + RedisKey key = Me(); + await db.KeyDeleteAsync(key); + await db.HashSetAsync(key, [new("old", 1), new("keep", 2)]); + + await using var fieldSet = HashImport.Create("name", "email", "age"); + await db.HashImportAsync(key, fieldSet, Values("alice", "a@x", 30)); + + // HIMPORT SET replaces the whole hash: the pre-existing 'old'/'keep' fields are gone + Assert.Equal(3, await db.HashLengthAsync(key)); + Assert.False(await db.HashExistsAsync(key, "old")); + Assert.Equal("alice", await db.HashGetAsync(key, "name")); + } + + [Fact] + public async Task MismatchedValueCount_ThrowsBeforeSending() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + await using var fieldSet = HashImport.Create("name", "email", "age"); + // 3 fields but only 1 value -> synchronous ArgumentException, before any Task is returned + Assert.Throws(() => { _ = db.HashImportAsync(Me(), fieldSet, new RedisValue[] { "only-one" }); }); + } + + [Fact] + public async Task NullFieldSet_Throws() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + Assert.Throws(() => { _ = db.HashImportAsync(Me(), null!, new RedisValue[] { "x" }); }); + } + + [Fact] + public async Task WrongTypeKey_ThrowsButOtherKeysSucceed() + { + 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 + + await using var fieldSet = HashImport.Create("name", "email", "age"); + + var ex = await Assert.ThrowsAsync(() => db.HashImportAsync(k0, fieldSet, Values("alice", "a@x", 30))); + Assert.StartsWith("WRONGTYPE", ex.Message); + + // each import is applied on its own: a later valid key still succeeds despite the earlier failure + await db.HashImportAsync(k1, fieldSet, Values("bob", "b@x", 25)); + Assert.Equal("bob", await db.HashGetAsync(k1, "name")); + Assert.Equal("i-am-a-string", await db.StringGetAsync(k0)); // untouched + } + + [Fact] + public void DuplicateFieldNames_RejectedAtCreate() + { + // rejected client-side: the server would reject the PREPARE, but that is injected fire-and-forget and would + // only surface indirectly as a "no such fieldset" failure on every SET - so we fail fast at the mistake. + var ex = Assert.Throws(() => HashImport.Create("f1", "f1")); + Assert.Contains("Duplicate field name", ex.Message); + } + + [Fact] + public async Task FieldsAreSnapshotAtCreate() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + RedisKey key = Me(); + await db.KeyDeleteAsync(key); + + var fieldNames = new RedisValue[] { "name", "email", "age" }; + await using var fieldSet = HashImport.Create(fieldNames); + fieldNames[0] = "MUTATED"; // mutate the caller's array after Create - must not affect the field-set + + await db.HashImportAsync(key, fieldSet, Values("alice", "a@x", 30)); + Assert.Equal("alice", await db.HashGetAsync(key, "name")); // still 'name', not 'MUTATED' + Assert.False(await db.HashExistsAsync(key, "MUTATED")); + } + + [Fact] + public void NullFieldName_RejectedAtCreate() + { + var ex = Assert.Throws(() => HashImport.Create("a", RedisValue.Null)); + Assert.Contains("null", ex.Message); + } + + [Fact] + public void EmptyFieldName_Allowed() + { + // the server accepts an empty field name (a hash can legitimately have an empty-string field), so we do too + using var fieldSet = HashImport.Create("", "b"); + Assert.NotNull(fieldSet); + } + + [Fact] + public async Task KeyPrefixIsolation_PrefixesKey() + { + 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 using var fieldSet = HashImport.Create("name", "email", "age"); + await db.HashImportAsync(inner, fieldSet, Values("alice", "a@x", 30)); + + // written under the prefixed key + Assert.Equal("alice", await raw.HashGetAsync(full, "name")); + } + + [Fact] + public async Task WorksInsideBatch() + { + 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]); + + await using var fieldSet = HashImport.Create("name", "email", "age"); + var batch = db.CreateBatch(); + var t1 = batch.HashImportAsync(k1, fieldSet, Values("alice", "a@x", 30)); + var t2 = batch.HashImportAsync(k2, fieldSet, Values("bob", "b@x", 25)); + batch.Execute(); + await Task.WhenAll(t1, t2); + + Assert.Equal("alice", await db.HashGetAsync(k1, "name")); + Assert.Equal("bob", await db.HashGetAsync(k2, "name")); + } + + [Fact] + public async Task UseAfterDispose_Throws() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + var fieldSet = HashImport.Create("name", "email", "age"); + fieldSet.Dispose(); + // rejected before anything is sent (the field-set may already have been DISCARDed on the server) + Assert.Throws(() => { _ = db.HashImportAsync(Me(), fieldSet, Values("a", "b", 1)); }); + } + + [Fact] + public async Task DoubleDispose_IsNoOp() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var db = conn.GetDatabase(); + RedisKey key = Me(); + await db.KeyDeleteAsync(key); + + var fieldSet = HashImport.Create("name", "email", "age"); + await db.HashImportAsync(key, fieldSet, Values("alice", "a@x", 30)); + fieldSet.Dispose(); + fieldSet.Dispose(); // idempotent: no second DISCARD, no throw + await fieldSet.DisposeAsync(); // also idempotent across the sync/async forms + + Assert.Equal("alice", await db.HashGetAsync(key, "name")); + } + + [Fact] + public async Task NotSupportedInsideTransaction() + { + await using var conn = Create(require: RedisFeatures.v8_10_0); + var tran = conn.GetDatabase().CreateTransaction(); + await using var fieldSet = HashImport.Create("name", "email", "age"); + // the transaction guard runs synchronously + Assert.Throws(() => { _ = tran.HashImportAsync(Me(), fieldSet, Values("a", "b", 1)); }); + } +} diff --git a/tests/StackExchange.Redis.Tests/HighIntegrityHashImportTests.cs b/tests/StackExchange.Redis.Tests/HighIntegrityHashImportTests.cs new file mode 100644 index 000000000..a2db70f7b --- /dev/null +++ b/tests/StackExchange.Redis.Tests/HighIntegrityHashImportTests.cs @@ -0,0 +1,13 @@ +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Reruns the full suite with high-integrity mode enabled, exercising the coexistence of +/// the fire-and-forget HIMPORT PREPARE injected into the write path (which is not high-integrity) with the +/// per-command high-integrity checksum framing applied to the following HIMPORT SETs on the same connection. +/// +public class HighIntegrityHashImportTests(ITestOutputHelper output, SharedConnectionFixture fixture) : HashImportTests(output, fixture) +{ + internal override bool HighIntegrity => true; +} diff --git a/tests/StackExchange.Redis.Tests/RoundTripUnitTests/HashImport.cs b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/HashImport.cs new file mode 100644 index 000000000..ac6cbbf57 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/HashImport.cs @@ -0,0 +1,75 @@ +using System; +using System.Text; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests.RoundTripUnitTests; + +/// +/// Verifies the exact wire bytes of the individual HIMPORT messages. The field-set name is written as an +/// 8-byte bulk string (the token's opaque id); it is derived from the live token here rather than hard-coded, since +/// the id is a process-wide monotonic counter. +/// +public class HashImport(ITestOutputHelper log) +{ + // RESP bulk-string encoding of an arbitrary byte payload (may contain non-printable bytes). + private static string Bulk(byte[] bytes) + { + var sb = new StringBuilder().Append('$').Append(bytes.Length).Append("\r\n"); + foreach (var b in bytes) sb.Append((char)b); + return sb.Append("\r\n").ToString(); + } + + [Fact(Timeout = 5000)] + public async Task Prepare_RoundTrips() + { + var token = StackExchange.Redis.HashImport.Create("name", "email", "age"); + byte[] name = BitConverter.GetBytes(token.Id); + var msg = new HashImportPrepareMessage(0, CommandFlags.None, token); + + // HIMPORT PREPARE name email age + var request = "*6\r\n$7\r\nHIMPORT\r\n$7\r\nPREPARE\r\n" + Bulk(name) + "$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() + { + var token = StackExchange.Redis.HashImport.Create("f1", "f2"); + byte[] name = BitConverter.GetBytes(token.Id); + ReadOnlyMemory values = new RedisValue[] { "v1", "v2" }; + var msg = new HashImportSetMessage(0, CommandFlags.None, token, (RedisKey)"user:1", values); + + // HIMPORT SET user:1 v1 v2 + var request = "*6\r\n$7\r\nHIMPORT\r\n$3\r\nSET\r\n$6\r\nuser:1\r\n" + Bulk(name) + "$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() + { + var token = StackExchange.Redis.HashImport.Create("only"); + byte[] name = BitConverter.GetBytes(token.Id); + ReadOnlyMemory values = new RedisValue[] { "v" }; + var msg = new HashImportSetMessage(0, CommandFlags.None, token, (RedisKey)"k", values); + + var request = "*5\r\n$7\r\nHIMPORT\r\n$3\r\nSET\r\n$1\r\nk\r\n" + Bulk(name) + "$1\r\nv\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 Discard_RoundTrips() + { + var token = StackExchange.Redis.HashImport.Create("f"); + byte[] name = BitConverter.GetBytes(token.Id); + var msg = new HashImportDiscardMessage(0, CommandFlags.None, token); + + // HIMPORT DISCARD + var request = "*3\r\n$7\r\nHIMPORT\r\n$7\r\nDISCARD\r\n" + Bulk(name); + var result = await TestConnection.ExecuteAsync(msg, ResultProcessor.DemandOK, request, "+OK\r\n", log: log); + Assert.True(result); + } +}