Add caller-owned KV directory queries

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<T, TKey>.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

Directory.Build.props

@@ -11,7 +11,7 @@
 
   <PropertyGroup Condition="'$(IsPackable)' != 'false'">
     <!-- Assembly identity stays fixed permanently; release only by advancing PackageVersion. -->
-    <PackageVersion>1.3.1</PackageVersion>
+    <PackageVersion>1.4.0</PackageVersion>
     <Version>$(PackageVersion)</Version>
     <AssemblyVersion>1.0.0.0</AssemblyVersion>
     <InformationalVersion>$(PackageVersion)</InformationalVersion>
@@ -26,7 +26,7 @@
     <DebugType>portable</DebugType>
     <PackageReadmeFile>README.md</PackageReadmeFile>
     <PackageTags>fitz distributed-systems kv queue rpc lease notice stream schedule</PackageTags>
-    <PackageReleaseNotes>Rebuilds Cntryl.Fitz.Extensions around bounded covering indexes, keyset cursors, explicit write costs, and resumable index generations. See CHANGELOG.md.</PackageReleaseNotes>
+    <PackageReleaseNotes>Adds transaction route reporting and caller-owned transaction queries for Cntryl.Fitz.Extensions. See CHANGELOG.md.</PackageReleaseNotes>
     <PackageOutputPath>$(MSBuildThisFileDirectory)artifacts\packages</PackageOutputPath>
   </PropertyGroup>
 

README.md

@@ -141,8 +141,16 @@ var teams = new KvDirectory<Team, Guid>(
 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

src/Fitz.Abstractions/Domains/Kv/IKvTransaction.cs

@@ -5,6 +5,13 @@ namespace Cntryl.Fitz;
 /// </summary>
 public interface IKvTransaction : IAsyncDisposable
 {
+    /// <summary>
+    /// Gets the exact KV route this transaction was opened on.
+    /// </summary>
+    /// <exception cref="NotSupportedException">The implementation does not report its route.</exception>
+    string Route => throw new NotSupportedException(
+        $"{GetType().FullName} does not report the route it was opened on. Implement IKvTransaction.Route.");
+
     /// <summary>
     /// Reads a key from the transaction snapshot.
     /// </summary>

src/Fitz.Core/Domains/Kv/KvTransaction.cs

@@ -46,6 +46,9 @@ internal KvTransaction(
         }
     }
 
+    /// <inheritdoc />
+    public string Route => _route;
+
     /// <inheritdoc />
     public async Task<KvGetResult> GetAsync(ReadOnlyMemory<byte> key, CancellationToken ct = default)
     {

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);
     }
 
-    /// <summary>Executes one bounded keyset query against a configured covering index.</summary>
+    /// <summary>
+    /// 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
+    /// <see cref="IKvTransaction.Route"/>, so a cursor issued for one route is rejected on another.
+    /// </summary>
+    /// <exception cref="NotSupportedException">The transaction does not report its route.</exception>
+    public async ValueTask<Page<T>> QueryAsync(
+        IKvTransaction transaction,
+        KvDirectoryQuery<T> query,
+        CancellationToken ct = default)
+    {
+        ArgumentNullException.ThrowIfNull(transaction);
+        ArgumentNullException.ThrowIfNull(query);
+        var plan = Plan(transaction.Route, query);
+        return await ExecuteAsync(transaction, plan, ct).ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
     [SuppressMessage("Reliability", "CA2007:Consider calling ConfigureAwait on the awaited task",
         Justification = "The await-using declaration must retain the transaction type.")]
     public async ValueTask<Page<T>> QueryAsync(
@@ -168,33 +188,10 @@ public async ValueTask<Page<T>> 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<KvPair>(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<T>(items, next);
+        return await ExecuteAsync(transaction, plan, ct).ConfigureAwait(false);
     }
 
     /// <summary>
@@ -363,6 +360,42 @@ T Deserialize(ReadOnlyMemory<byte> value) =>
         JsonSerializer.Deserialize(value.Span, _valueTypeInfo)
         ?? throw new InvalidOperationException("The stored directory entry could not be deserialized.");
 
+    QueryPlan Plan(string route, KvDirectoryQuery<T> 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<Page<T>> ExecuteAsync(IKvTransaction transaction, QueryPlan plan, CancellationToken ct)
+    {
+        var matches = new List<KvPair>(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<T>(items, next);
+    }
+
+    readonly record struct QueryPlan(KvScanQuery Scan, byte[] Fingerprint, int Limit);
+
     byte[] Fingerprint(
         string route,
         KvDirectoryIndex<T> index,

src/Fitz.Testing/InMemoryKvClient.cs

@@ -591,6 +591,8 @@ internal InMemoryKvTransaction(
             _scanPageSize = scanPageSize;
         }
 
+        public string Route => _route;
+
         public Task<KvGetResult> GetAsync(ReadOnlyMemory<byte> key, CancellationToken ct = default)
         {
             ct.ThrowIfCancellationRequested();

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<ReadOnlyMemory<byte>>(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)]

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<KvDirectoryQueryException>(() => 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<KvDirectoryQueryException>(() => 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<ArgumentNullException>(() =>
+            Directory.QueryAsync((IKvTransaction)null!, ByNameV1.Query()).AsTask());
+        var missingQuery = await Assert.ThrowsAsync<ArgumentNullException>(() =>
+            Directory.QueryAsync(transaction, null!).AsTask());
+
+        // Assert
+        Assert.Equal("transaction", missingTransaction.ParamName);
+        Assert.Equal("query", missingQuery.ParamName);
+    }
+
     [Fact]
     public async Task ShouldReplaceWithoutReadingGivenPreviousValue()
     {

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()
     {

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<NotSupportedException>(() => 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<KvScanResult> ScanAsync(KvScanQuery query, CancellationToken ct = default) =>