From 5798a5804eab915174551d33ed28a04525af519d Mon Sep 17 00:00:00 2001 From: Jeff Repanich Date: Fri, 18 Sep 2026 13:27:53 -0400 Subject: [PATCH] Add caller-owned KV directory queries --- CHANGELOG.md | 22 +++++ Directory.Build.props | 4 +- README.md | 8 ++ .../Domains/Kv/IKvTransaction.cs | 7 ++ src/Fitz.Core/Domains/Kv/KvTransaction.cs | 3 + src/Fitz.Extensions/KvDirectory.cs | 85 ++++++++++++----- src/Fitz.Testing/InMemoryKvClient.cs | 2 + test/Fitz.Core.Tests/Unit/KvClientTests.cs | 19 ++++ .../Fitz.Extensions.Tests/KvDirectoryTests.cs | 95 +++++++++++++++++++ .../InMemoryKvClientTests.cs | 15 +++ test/Fitz.Testing.Tests/KvScanHelpersTests.cs | 14 +++ 11 files changed, 246 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c2056ca..6dbc92b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,28 @@ never breaks for consumers. `test/Fitz.PackageConsumer` asserts this on every CI ## [Unreleased] +## [1.4.0] - 2026-09-18 + +### Added + +- `IKvTransaction.Route` reports the exact route a transaction was opened on. The Fitz + transaction and `Cntryl.Fitz.Testing`'s `InMemoryKvClient` implement it. It is a default + interface member, so existing third-party implementations still compile, but they throw + `NotSupportedException` from it until they implement it. +- **`Cntryl.Fitz.Extensions`:** `KvDirectory.QueryAsync(IKvTransaction, query, ct)` runs + a query through a caller-owned transaction, as `GetAsync` and the write operations already do. + A read-write transaction sees its own staged writes. Cursors bind to `IKvTransaction.Route`, so + they interoperate with the `(client, route)` overload and are rejected on any other route. The + `(client, route)` overload now delegates to the same query logic and still validates the query + and cursor before opening its transaction. + +## [1.3.1] - 2026-09-18 + +### Fixed + +- Encode the KV `SCAN` limit as the broker's unsigned 32-bit wire value instead of an unsigned + 64-bit value, preventing the following fields from being shifted and misread. + ## [1.3.0] - 2026-09-17 ### Changed diff --git a/Directory.Build.props b/Directory.Build.props index 0851144..b3d7c3a 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -11,7 +11,7 @@ - 1.3.1 + 1.4.0 $(PackageVersion) 1.0.0.0 $(PackageVersion) @@ -26,7 +26,7 @@ portable README.md fitz distributed-systems kv queue rpc lease notice stream schedule - Rebuilds Cntryl.Fitz.Extensions around bounded covering indexes, keyset cursors, explicit write costs, and resumable index generations. See CHANGELOG.md. + Adds transaction route reporting and caller-owned transaction queries for Cntryl.Fitz.Extensions. See CHANGELOG.md. $(MSBuildThisFileDirectory)artifacts\packages diff --git a/README.md b/README.md index 22f4ece..336c709 100644 --- a/README.md +++ b/README.md @@ -141,8 +141,16 @@ var teams = new KvDirectory( await teams.InsertAsync(transaction, team, ct); // no read var page = await teams.QueryAsync(client, route, byName.Query().Take(50).After(cursor), ct); + +// Or through a transaction the caller already holds; a read-write one sees its staged writes. +var same = await teams.QueryAsync(transaction, byName.Query().Take(50).After(cursor), ct); ``` +Record operations and queries accept a caller-owned `IKvTransaction`; the `(client, route)` +read forms are shorthand that open a short read-only transaction. Cursors are bound to the route +the query ran on, taken from `IKvTransaction.Route`, so a cursor from one route is rejected on +another. + `UpsertAsync` reads the previous primary record so it can remove old index rows. Hot paths that already know the previous value use `ReplaceAsync(previous, current)` instead. To add an index generation, deploy a schema containing both generations (ordinary writes then dual-write), backfill diff --git a/src/Fitz.Abstractions/Domains/Kv/IKvTransaction.cs b/src/Fitz.Abstractions/Domains/Kv/IKvTransaction.cs index 5c368b8..3a206a4 100644 --- a/src/Fitz.Abstractions/Domains/Kv/IKvTransaction.cs +++ b/src/Fitz.Abstractions/Domains/Kv/IKvTransaction.cs @@ -5,6 +5,13 @@ namespace Cntryl.Fitz; /// public interface IKvTransaction : IAsyncDisposable { + /// + /// Gets the exact KV route this transaction was opened on. + /// + /// The implementation does not report its route. + string Route => throw new NotSupportedException( + $"{GetType().FullName} does not report the route it was opened on. Implement IKvTransaction.Route."); + /// /// Reads a key from the transaction snapshot. /// diff --git a/src/Fitz.Core/Domains/Kv/KvTransaction.cs b/src/Fitz.Core/Domains/Kv/KvTransaction.cs index c59682a..1706413 100644 --- a/src/Fitz.Core/Domains/Kv/KvTransaction.cs +++ b/src/Fitz.Core/Domains/Kv/KvTransaction.cs @@ -46,6 +46,9 @@ internal KvTransaction( } } + /// + public string Route => _route; + /// public async Task GetAsync(ReadOnlyMemory key, CancellationToken ct = default) { diff --git a/src/Fitz.Extensions/KvDirectory.cs b/src/Fitz.Extensions/KvDirectory.cs index 31a15e0..ee0934d 100644 --- a/src/Fitz.Extensions/KvDirectory.cs +++ b/src/Fitz.Extensions/KvDirectory.cs @@ -156,7 +156,27 @@ public async ValueTask DeleteAsync(IKvTransaction transaction, TKey identity, Ca return await GetAsync(transaction, identity, ct).ConfigureAwait(false); } - /// Executes one bounded keyset query against a configured covering index. + /// + /// Executes one bounded keyset query against a configured covering index through a caller-owned + /// transaction. A read-write transaction sees its own staged writes. Cursors are bound to + /// , so a cursor issued for one route is rejected on another. + /// + /// The transaction does not report its route. + public async ValueTask> QueryAsync( + IKvTransaction transaction, + KvDirectoryQuery query, + CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(transaction); + ArgumentNullException.ThrowIfNull(query); + var plan = Plan(transaction.Route, query); + return await ExecuteAsync(transaction, plan, ct).ConfigureAwait(false); + } + + /// + /// Executes one bounded keyset query against a configured covering index in a short read-only + /// transaction. The query and cursor are validated before the transaction is opened. + /// [SuppressMessage("Reliability", "CA2007:Consider calling ConfigureAwait on the awaited task", Justification = "The await-using declaration must retain the transaction type.")] public async ValueTask> QueryAsync( @@ -168,33 +188,10 @@ public async ValueTask> QueryAsync( ArgumentNullException.ThrowIfNull(client); ArgumentException.ThrowIfNullOrWhiteSpace(route); ArgumentNullException.ThrowIfNull(query); - var index = Resolve(query.Index); - var limit = query.Limit ?? _options.DefaultPageSize; - ValidateLimit(limit); - var prefix = IndexPrefix(index, query.Prefix); - var rangeStart = LexKey.EncodeFirst(prefix).AsMemory(); - var rangeEnd = LexKey.EncodeLast(prefix).AsMemory(); - var fingerprint = Fingerprint(route, index, query.IsDescending, query.Prefix, "query"); - var cursorKey = DecodeCursor(query.Cursor, fingerprint); - ValidateCursorRange(cursorKey, rangeStart, rangeEnd); - var scan = query.IsDescending - ? new KvScanQuery(rangeStart, cursorKey ?? rangeEnd, (uint)(limit + 1), Reverse: true) - : new KvScanQuery(cursorKey is null ? rangeStart : After(cursorKey.Value.Span), rangeEnd, - (uint)(limit + 1)); + var plan = Plan(route, query); await using var transaction = await client.BeginAsync(route, KvDurability.Async, KvMode.ReadOnly, ct) .ConfigureAwait(false); - var matches = new List(limit + 1); - await foreach (var pair in transaction.ScanAllAsync(scan, ct).ConfigureAwait(false)) - { - matches.Add(pair); - if (matches.Count > limit) - break; - } - var hasMore = matches.Count > limit; - var returned = matches.Take(limit).ToArray(); - var items = returned.Select(pair => Deserialize(pair.Value)).ToArray(); - var next = hasMore ? EncodeCursor(fingerprint, returned[^1].Key) : null; - return new Page(items, next); + return await ExecuteAsync(transaction, plan, ct).ConfigureAwait(false); } /// @@ -363,6 +360,42 @@ T Deserialize(ReadOnlyMemory value) => JsonSerializer.Deserialize(value.Span, _valueTypeInfo) ?? throw new InvalidOperationException("The stored directory entry could not be deserialized."); + QueryPlan Plan(string route, KvDirectoryQuery query) + { + var index = Resolve(query.Index); + var limit = query.Limit ?? _options.DefaultPageSize; + ValidateLimit(limit); + var prefix = IndexPrefix(index, query.Prefix); + var rangeStart = LexKey.EncodeFirst(prefix).AsMemory(); + var rangeEnd = LexKey.EncodeLast(prefix).AsMemory(); + var fingerprint = Fingerprint(route, index, query.IsDescending, query.Prefix, "query"); + var cursorKey = DecodeCursor(query.Cursor, fingerprint); + ValidateCursorRange(cursorKey, rangeStart, rangeEnd); + var scan = query.IsDescending + ? new KvScanQuery(rangeStart, cursorKey ?? rangeEnd, (uint)(limit + 1), Reverse: true) + : new KvScanQuery(cursorKey is null ? rangeStart : After(cursorKey.Value.Span), rangeEnd, + (uint)(limit + 1)); + return new QueryPlan(scan, fingerprint, limit); + } + + async ValueTask> ExecuteAsync(IKvTransaction transaction, QueryPlan plan, CancellationToken ct) + { + var matches = new List(plan.Limit + 1); + await foreach (var pair in transaction.ScanAllAsync(plan.Scan, ct).ConfigureAwait(false)) + { + matches.Add(pair); + if (matches.Count > plan.Limit) + break; + } + var hasMore = matches.Count > plan.Limit; + var returned = matches.Take(plan.Limit).ToArray(); + var items = returned.Select(pair => Deserialize(pair.Value)).ToArray(); + var next = hasMore ? EncodeCursor(plan.Fingerprint, returned[^1].Key) : null; + return new Page(items, next); + } + + readonly record struct QueryPlan(KvScanQuery Scan, byte[] Fingerprint, int Limit); + byte[] Fingerprint( string route, KvDirectoryIndex index, diff --git a/src/Fitz.Testing/InMemoryKvClient.cs b/src/Fitz.Testing/InMemoryKvClient.cs index 9888588..45e0f4d 100644 --- a/src/Fitz.Testing/InMemoryKvClient.cs +++ b/src/Fitz.Testing/InMemoryKvClient.cs @@ -591,6 +591,8 @@ internal InMemoryKvTransaction( _scanPageSize = scanPageSize; } + public string Route => _route; + public Task GetAsync(ReadOnlyMemory key, CancellationToken ct = default) { ct.ThrowIfCancellationRequested(); diff --git a/test/Fitz.Core.Tests/Unit/KvClientTests.cs b/test/Fitz.Core.Tests/Unit/KvClientTests.cs index 908e062..7533ad5 100644 --- a/test/Fitz.Core.Tests/Unit/KvClientTests.cs +++ b/test/Fitz.Core.Tests/Unit/KvClientTests.cs @@ -73,6 +73,25 @@ public async Task ShouldRejectInvalidFoundFlagGivenMalformedResponseWhenGettingV Assert.Equal("GET_INVALID_RESPONSE", error.Code); } + [Fact] + public async Task ShouldReportOpenedRouteGivenBegunTransaction() + { + // Arrange + using var kv = new KvClient((_, _, _) => + { + using var writer = new BinaryBufferWriter(); + writer.WriteU8(0); + writer.WriteU64(7); + return ValueTask.FromResult>(writer.Build()); + }); + + // Act + await using var transaction = await kv.BeginAsync("kv://prod/app/data", KvDurability.Sync, KvMode.ReadOnly); + + // Assert + Assert.Equal("kv://prod/app/data", transaction.Route); + } + [Theory] [InlineData(2, 0)] [InlineData(0, 260)] diff --git a/test/Fitz.Extensions.Tests/KvDirectoryTests.cs b/test/Fitz.Extensions.Tests/KvDirectoryTests.cs index 7728cb8..7b43274 100644 --- a/test/Fitz.Extensions.Tests/KvDirectoryTests.cs +++ b/test/Fitz.Extensions.Tests/KvDirectoryTests.cs @@ -237,6 +237,101 @@ public async Task ShouldRejectLimitGivenConfiguredMaximumExceeded() Assert.Equal(KvDirectoryQueryError.InvalidLimit, error.Kind); } + [Fact] + public async Task ShouldReturnSamePagesGivenCallerOwnedTransaction() + { + // Arrange + var client = new InMemoryKvClient(); + foreach (var name in new[] { "Alpha", "Beta", "Gamma" }) + await WriteAsync(client, transaction => Directory.InsertAsync( + transaction, new Widget(Guid.NewGuid(), name, 1))); + var expectedFirst = await Directory.QueryAsync(client, Route, ByNameV1.Query().Take(2)); + var expectedSecond = await Directory.QueryAsync( + client, Route, ByNameV1.Query().Take(2).After(expectedFirst.NextCursor)); + + // Act + await using var transaction = await client.BeginAsync(Route, KvDurability.Async, KvMode.ReadOnly); + var first = await Directory.QueryAsync(transaction, ByNameV1.Query().Take(2)); + var second = await Directory.QueryAsync(transaction, ByNameV1.Query().Take(2).After(first.NextCursor)); + + // Assert + Assert.Equal(expectedFirst.Items, first.Items); + Assert.Equal(expectedFirst.NextCursor, first.NextCursor); + Assert.Equal(expectedSecond.Items, second.Items); + Assert.Null(second.NextCursor); + } + + [Fact] + public async Task ShouldRejectCursorGivenCallerOwnedTransactionOnDifferentRoute() + { + // Arrange + var client = new InMemoryKvClient(); + await WriteAsync(client, transaction => Directory.InsertAsync( + transaction, new Widget(Guid.NewGuid(), "Alpha", 1))); + await WriteAsync(client, transaction => Directory.InsertAsync( + transaction, new Widget(Guid.NewGuid(), "Beta", 2))); + var first = await Directory.QueryAsync(client, Route, ByNameV1.Query().Take(1)); + await using var other = await client.BeginAsync(OtherRoute, KvDurability.Async, KvMode.ReadOnly); + + // Act + var error = await Assert.ThrowsAsync(() => Directory.QueryAsync( + other, ByNameV1.Query().Take(1).After(first.NextCursor)).AsTask()); + + // Assert + Assert.Equal(KvDirectoryQueryError.CursorMismatch, error.Kind); + } + + [Fact] + public async Task ShouldNotBeginTransactionGivenInvalidCursorOnClientQuery() + { + // Arrange + var client = new InMemoryKvClient(); + client.ClearOperations(); + + // Act + var error = await Assert.ThrowsAsync(() => Directory.QueryAsync( + client, Route, ByNameV1.Query().After("not-a-cursor")).AsTask()); + + // Assert + Assert.Equal(KvDirectoryQueryError.InvalidCursor, error.Kind); + Assert.DoesNotContain(client.Operations, static operation => operation.Operation is KvTestOperation.Begin); + } + + [Fact] + public async Task ShouldSeeStagedWritesGivenQueryInsideReadWriteTransaction() + { + // Arrange + var client = new InMemoryKvClient(); + await WriteAsync(client, transaction => Directory.InsertAsync( + transaction, new Widget(Guid.NewGuid(), "Alpha", 1))); + await using var transaction = await client.BeginAsync(Route, KvDurability.Async, KvMode.ReadWrite); + await Directory.InsertAsync(transaction, new Widget(Guid.NewGuid(), "Beta", 2)); + + // Act + var page = await Directory.QueryAsync(transaction, ByNameV1.Query()); + + // Assert + Assert.Equal(["Alpha", "Beta"], page.Items.Select(static widget => widget.Name)); + } + + [Fact] + public async Task ShouldRejectQueryGivenMissingTransactionOrQuery() + { + // Arrange + var client = new InMemoryKvClient(); + await using var transaction = await client.BeginAsync(Route, KvDurability.Async, KvMode.ReadOnly); + + // Act + var missingTransaction = await Assert.ThrowsAsync(() => + Directory.QueryAsync((IKvTransaction)null!, ByNameV1.Query()).AsTask()); + var missingQuery = await Assert.ThrowsAsync(() => + Directory.QueryAsync(transaction, null!).AsTask()); + + // Assert + Assert.Equal("transaction", missingTransaction.ParamName); + Assert.Equal("query", missingQuery.ParamName); + } + [Fact] public async Task ShouldReplaceWithoutReadingGivenPreviousValue() { diff --git a/test/Fitz.Testing.Tests/InMemoryKvClientTests.cs b/test/Fitz.Testing.Tests/InMemoryKvClientTests.cs index a0b04b6..a2ede8f 100644 --- a/test/Fitz.Testing.Tests/InMemoryKvClientTests.cs +++ b/test/Fitz.Testing.Tests/InMemoryKvClientTests.cs @@ -6,6 +6,21 @@ public sealed class InMemoryKvClientTests { const string Route = "kv://tenant/app/entities"; + [Theory] + [InlineData(KvMode.ReadOnly)] + [InlineData(KvMode.ReadWrite)] + public async Task ShouldReportOpenedRouteGivenBegunTransaction(KvMode mode) + { + // Arrange + var client = new InMemoryKvClient(); + + // Act + await using var transaction = await client.BeginAsync(Route, KvDurability.Async, mode); + + // Assert + Assert.Equal(Route, transaction.Route); + } + [Fact] public async Task ShouldExposeOwnWritesAndPersistOnlyCommittedChangesGivenTransactionLifecycle() { diff --git a/test/Fitz.Testing.Tests/KvScanHelpersTests.cs b/test/Fitz.Testing.Tests/KvScanHelpersTests.cs index 5c8b61e..61b0bd3 100644 --- a/test/Fitz.Testing.Tests/KvScanHelpersTests.cs +++ b/test/Fitz.Testing.Tests/KvScanHelpersTests.cs @@ -35,6 +35,20 @@ public async Task ShouldResumeAfterExactLastKeyGivenForwardScanAll() Assert.Equal(new byte[] { 0x10, 0xFF, 0x00 }, transaction.Queries[1].StartKey!.Value.ToArray()); } + [Fact] + public async Task ShouldFailLoudlyGivenImplementationThatDoesNotReportRoute() + { + // Arrange + await using IKvTransaction transaction = new EmptyContinuationTransaction(); + + // Act + var error = Assert.Throws(() => transaction.Route); + + // Assert + Assert.Contains(nameof(EmptyContinuationTransaction), error.Message, StringComparison.Ordinal); + Assert.Contains("IKvTransaction.Route", error.Message, StringComparison.Ordinal); + } + sealed class EmptyContinuationTransaction : IKvTransaction { public Task ScanAsync(KvScanQuery query, CancellationToken ct = default) =>