Add bounded primary directory paging

CHANGELOG.md

@@ -16,6 +16,16 @@ never breaks for consumers. `test/Fitz.PackageConsumer` asserts this on every CI
 
 ## [Unreleased]
 
+## [1.4.1] - 2026-09-21
+
+### Added
+
+- **`Cntryl.Fitz.Extensions`:** `KvDirectory<T, TKey>.QueryPrimaryAsync` pages existing primary
+  records in stable key order through either a caller-owned transaction or a short read-only
+  transaction. Each request is bounded to `limit + 1`; opaque cursors bind to the route,
+  directory, and primary-query shape. Existing records require no secondary index, backfill,
+  migration marker, or request-path write.
+
 ## [1.4.0] - 2026-09-18
 
 ### Added

Directory.Build.props

@@ -11,7 +11,7 @@
 
   <PropertyGroup Condition="'$(IsPackable)' != 'false'">
     <!-- Assembly identity stays fixed permanently; release only by advancing PackageVersion. -->
-    <PackageVersion>1.4.0</PackageVersion>
+    <PackageVersion>1.4.1</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>Adds transaction route reporting and caller-owned transaction queries for Cntryl.Fitz.Extensions. See CHANGELOG.md.</PackageReleaseNotes>
+    <PackageReleaseNotes>Adds bounded read-only primary-key paging for Cntryl.Fitz.Extensions directories. See CHANGELOG.md.</PackageReleaseNotes>
     <PackageOutputPath>$(MSBuildThisFileDirectory)artifacts\packages</PackageOutputPath>
   </PropertyGroup>
 

README.md

@@ -122,9 +122,9 @@ key-schema neutral. Use `Cntryl.LexKey` to construct typed keys and range bounds
 
 ## Indexed KV directories
 
-`Cntryl.Fitz.Extensions` turns declared query shapes into bounded covering-index scans. The
-application owns which indexes exist and maps external sort or filter input to those handles;
-Fitz owns atomic index maintenance, bounds, keyset continuation, and generation migration.
+`Cntryl.Fitz.Extensions` turns declared query shapes into bounded primary-key or covering-index
+scans. The application owns which indexes exist and maps external sort or filter input to those
+handles; Fitz owns atomic index maintenance, bounds, keyset continuation, and generation migration.
 
 ```csharp
 var byName = new KvDirectoryIndex<Team>(
@@ -144,13 +144,20 @@ var page = await teams.QueryAsync(client, route,
 
 // 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);
+
+// Page the existing primary rows directly when identity order is the required order.
+var primaryPage = await teams.QueryPrimaryAsync(transaction, limit: 50, cursor: primaryCursor, 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.
 
+Primary-key pages are ascending and bounded, and use their own route-bound opaque cursors.
+They read the primary rows already maintained by the directory, including historical rows, so
+they require no secondary index, backfill, migration marker, or request-path write.
+
 `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.Extensions/KvDirectory.cs

@@ -194,6 +194,55 @@ public async ValueTask<Page<T>> QueryAsync(
         return await ExecuteAsync(transaction, plan, ct).ConfigureAwait(false);
     }
 
+    /// <summary>
+    /// Reads one bounded page in primary-key order through a caller-owned transaction. A read-write
+    /// transaction sees its own staged writes. Cursors are bound to <see cref="IKvTransaction.Route"/>
+    /// and this directory's primary rows, so a cursor issued for another route or query is rejected.
+    /// </summary>
+    /// <param name="transaction">The caller-owned transaction used for the scan.</param>
+    /// <param name="limit">The maximum number of primary records to return.</param>
+    /// <param name="cursor">An opaque continuation cursor from a previous primary page.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>Primary records in ascending key order and a cursor for the next page, if any.</returns>
+    /// <exception cref="NotSupportedException">The transaction does not report its route.</exception>
+    public async ValueTask<Page<T>> QueryPrimaryAsync(
+        IKvTransaction transaction,
+        int limit,
+        string? cursor = null,
+        CancellationToken ct = default)
+    {
+        ArgumentNullException.ThrowIfNull(transaction);
+        var plan = PlanPrimary(transaction.Route, limit, cursor);
+        return await ExecuteAsync(transaction, plan, ct).ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Reads one bounded page in primary-key order through a short read-only transaction. The limit
+    /// and cursor are validated before the transaction is opened.
+    /// </summary>
+    /// <param name="client">The KV client used to open the read-only transaction.</param>
+    /// <param name="route">The exact KV route containing the directory.</param>
+    /// <param name="limit">The maximum number of primary records to return.</param>
+    /// <param name="cursor">An opaque continuation cursor from a previous primary page.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>Primary records in ascending key order and a cursor for the next page, if any.</returns>
+    [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>> QueryPrimaryAsync(
+        IKvClient client,
+        string route,
+        int limit,
+        string? cursor = null,
+        CancellationToken ct = default)
+    {
+        ArgumentNullException.ThrowIfNull(client);
+        ArgumentException.ThrowIfNullOrWhiteSpace(route);
+        var plan = PlanPrimary(route, limit, cursor);
+        await using var transaction = await client.BeginAsync(route, KvDurability.Async, KvMode.ReadOnly, ct)
+            .ConfigureAwait(false);
+        return await ExecuteAsync(transaction, plan, ct).ConfigureAwait(false);
+    }
+
     /// <summary>
     /// Backfills one configured index generation from stable primary records in a committed,
     /// resumable batch. Deploy the upgraded schema first so ordinary writes dual-write old and new generations.
@@ -378,6 +427,20 @@ QueryPlan Plan(string route, KvDirectoryQuery<T> query)
         return new QueryPlan(scan, fingerprint, limit);
     }
 
+    QueryPlan PlanPrimary(string route, int limit, string? cursor)
+    {
+        ValidateLimit(limit);
+        var primaryPrefix = PrimaryPrefix();
+        var rangeStart = LexKey.EncodeFirst(primaryPrefix).AsMemory();
+        var rangeEnd = LexKey.EncodeLast(primaryPrefix).AsMemory();
+        var fingerprint = PrimaryFingerprint(route);
+        var cursorKey = DecodeCursor(cursor, fingerprint);
+        ValidateCursorRange(cursorKey, rangeStart, rangeEnd);
+        var scan = 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);
@@ -416,6 +479,15 @@ byte[] Fingerprint(
         return hash.GetHashAndReset()[..FingerprintLength];
     }
 
+    byte[] PrimaryFingerprint(string route)
+    {
+        using var hash = IncrementalHash.CreateHash(HashAlgorithmName.SHA256);
+        Append(hash, "primary-query");
+        Append(hash, route);
+        Append(hash, _name);
+        return hash.GetHashAndReset()[..FingerprintLength];
+    }
+
     static void Append(IncrementalHash hash, string value)
     {
         var bytes = Encoding.UTF8.GetBytes(value);
@@ -481,7 +553,7 @@ static void ValidateCursorRange(
         if (cursorKey is not { } key)
             return;
         if (key.Span.SequenceCompareTo(rangeStart.Span) < 0 || key.Span.SequenceCompareTo(rangeEnd.Span) >= 0)
-            throw InvalidCursor("The cursor key falls outside the selected index range.");
+            throw InvalidCursor("The cursor key falls outside the selected query range.");
     }
 
     static byte[] After(ReadOnlySpan<byte> key)

src/Fitz.Extensions/Page.cs

@@ -2,7 +2,7 @@ namespace Cntryl.Fitz.Extensions;
 
 /// <summary>
 /// One keyset-ordered directory page. <see cref="NextCursor"/> is opaque and bound to the route,
-/// directory, index generation, direction, and prefix that produced it.
+/// directory, and primary or index query shape that produced it.
 /// </summary>
 /// <typeparam name="T">The item type.</typeparam>
 /// <param name="Items">The items in this page, in the query's effective order.</param>

test/Fitz.Extensions.Tests/KvDirectoryTests.cs

@@ -332,6 +332,177 @@ public async Task ShouldRejectQueryGivenMissingTransactionOrQuery()
         Assert.Equal("query", missingQuery.ParamName);
     }
 
+    [Fact]
+    public async Task ShouldPageHistoricalAndCurrentRowsGivenPrimaryKeysWhenQuerying()
+    {
+        // Arrange
+        var client = new InMemoryKvClient(new InMemoryKvClientOptions { ScanPageSize = 1 });
+        var historicalDirectory = CreateDirectory();
+        var currentDirectory = CreateDirectory(ByNameV1);
+        var historicalFirst = new Widget(
+            Guid.Parse("00000000-0000-0000-0000-000000000001"), "Historical one", 1);
+        var historicalSecond = new Widget(
+            Guid.Parse("00000000-0000-0000-0000-000000000002"), "Historical two", 2);
+        var current = new Widget(
+            Guid.Parse("00000000-0000-0000-0000-000000000003"), "Current", 3);
+        await WriteAsync(client, transaction => historicalDirectory.InsertAsync(transaction, historicalFirst));
+        await WriteAsync(client, transaction => historicalDirectory.InsertAsync(transaction, historicalSecond));
+        await WriteAsync(client, transaction => currentDirectory.InsertAsync(transaction, current));
+        await using var transaction = await client.BeginAsync(Route, KvDurability.Async, KvMode.ReadOnly);
+        client.ClearOperations();
+
+        // Act
+        var first = await currentDirectory.QueryPrimaryAsync(transaction, limit: 2);
+        var second = await currentDirectory.QueryPrimaryAsync(transaction, limit: 2, cursor: first.NextCursor);
+
+        // Assert
+        Assert.Equal([historicalFirst, historicalSecond], first.Items);
+        Assert.NotNull(first.NextCursor);
+        Assert.Equal([current], second.Items);
+        Assert.Null(second.NextCursor);
+        Assert.All(client.Operations, static operation => Assert.Equal(KvTestOperation.Scan, operation.Operation));
+        Assert.All(client.Operations, static operation => Assert.Equal((uint)3, operation.ScanQuery!.Limit));
+    }
+
+    [Fact]
+    public async Task ShouldUseReadOnlyTransactionGivenClientWhenQueryingPrimaryKeys()
+    {
+        // Arrange
+        var client = new InMemoryKvClient();
+        var widget = new Widget(Guid.NewGuid(), "Alpha", 1);
+        await WriteAsync(client, transaction => Directory.InsertAsync(transaction, widget));
+        client.ClearOperations();
+
+        // Act
+        var page = await Directory.QueryPrimaryAsync(client, Route, limit: 1);
+
+        // Assert
+        Assert.Equal(widget, Assert.Single(page.Items));
+        var begin = Assert.Single(client.Operations,
+            static operation => operation.Operation is KvTestOperation.Begin);
+        Assert.Equal(KvMode.ReadOnly, begin.Mode);
+        Assert.DoesNotContain(client.Operations, static operation => operation.Operation is
+            KvTestOperation.Get or KvTestOperation.Put or KvTestOperation.Insert or KvTestOperation.Delete or
+            KvTestOperation.DeleteRange or KvTestOperation.Commit);
+    }
+
+    [Fact]
+    public async Task ShouldRejectPrimaryCursorGivenDifferentRouteOrQueryWhenResuming()
+    {
+        // Arrange
+        var client = new InMemoryKvClient();
+        await WriteAsync(client, transaction => Directory.InsertAsync(
+            transaction, new Widget(Guid.Parse("00000000-0000-0000-0000-000000000001"), "Alpha", 1)));
+        await WriteAsync(client, transaction => Directory.InsertAsync(
+            transaction, new Widget(Guid.Parse("00000000-0000-0000-0000-000000000002"), "Beta", 2)));
+        await using var transaction = await client.BeginAsync(Route, KvDurability.Async, KvMode.ReadOnly);
+        var first = await Directory.QueryPrimaryAsync(transaction, limit: 1);
+        await using var other = await client.BeginAsync(OtherRoute, KvDurability.Async, KvMode.ReadOnly);
+
+        // Act
+        var routeError = await Assert.ThrowsAsync<KvDirectoryQueryException>(() =>
+            Directory.QueryPrimaryAsync(other, limit: 1, cursor: first.NextCursor).AsTask());
+        var queryError = await Assert.ThrowsAsync<KvDirectoryQueryException>(() => Directory.QueryAsync(
+            transaction, ByNameV1.Query().Take(1).After(first.NextCursor)).AsTask());
+
+        // Assert
+        Assert.Equal(KvDirectoryQueryError.CursorMismatch, routeError.Kind);
+        Assert.Equal(KvDirectoryQueryError.CursorMismatch, queryError.Kind);
+    }
+
+    [Theory]
+    [InlineData(0, null, KvDirectoryQueryError.InvalidLimit)]
+    [InlineData(4, null, KvDirectoryQueryError.InvalidLimit)]
+    [InlineData(1, "not-base64", KvDirectoryQueryError.InvalidCursor)]
+    public async Task ShouldRejectInvalidInputGivenPrimaryQueryWhenPlanning(
+        int limit,
+        string? cursor,
+        KvDirectoryQueryError expected)
+    {
+        // Arrange
+        var client = new InMemoryKvClient();
+        await using var transaction = await client.BeginAsync(Route, KvDurability.Async, KvMode.ReadOnly);
+        client.ClearOperations();
+
+        // Act
+        var error = await Assert.ThrowsAsync<KvDirectoryQueryException>(() =>
+            Directory.QueryPrimaryAsync(transaction, limit, cursor).AsTask());
+
+        // Assert
+        Assert.Equal(expected, error.Kind);
+        Assert.Empty(client.Operations);
+    }
+
+    [Fact]
+    public async Task ShouldRejectOversizedCursorGivenClientWhenPlanningPrimaryQuery()
+    {
+        // Arrange
+        var client = new InMemoryKvClient();
+
+        // Act
+        var error = await Assert.ThrowsAsync<KvDirectoryQueryException>(() =>
+            Directory.QueryPrimaryAsync(client, Route, limit: 1, cursor: new string('A', 5000)).AsTask());
+
+        // Assert
+        Assert.Equal(KvDirectoryQueryError.InvalidCursor, error.Kind);
+        Assert.Empty(client.Operations);
+    }
+
+    [Fact]
+    public async Task ShouldRejectCursorKeyOutsidePrimaryRangeGivenModifiedCursorWhenResuming()
+    {
+        // Arrange
+        var client = new InMemoryKvClient();
+        await WriteAsync(client, transaction => Directory.InsertAsync(
+            transaction, new Widget(Guid.Parse("00000000-0000-0000-0000-000000000001"), "Alpha", 1)));
+        await WriteAsync(client, transaction => Directory.InsertAsync(
+            transaction, new Widget(Guid.Parse("00000000-0000-0000-0000-000000000002"), "Beta", 2)));
+        await using var transaction = await client.BeginAsync(Route, KvDurability.Async, KvMode.ReadOnly);
+        var first = await Directory.QueryPrimaryAsync(transaction, limit: 1);
+        var payload = Convert.FromBase64String(first.NextCursor!);
+        payload[21] = 0;
+        client.ClearOperations();
+
+        // Act
+        var error = await Assert.ThrowsAsync<KvDirectoryQueryException>(() => Directory.QueryPrimaryAsync(
+            transaction, limit: 1, cursor: Convert.ToBase64String(payload)).AsTask());
+
+        // Assert
+        Assert.Equal(KvDirectoryQueryError.InvalidCursor, error.Kind);
+        Assert.Empty(client.Operations);
+    }
+
+    [Fact]
+    public async Task ShouldRejectMissingTransactionGivenPrimaryQueryWhenPlanning()
+    {
+        // Arrange
+        // Act
+        var error = await Assert.ThrowsAsync<ArgumentNullException>(() =>
+            Directory.QueryPrimaryAsync((IKvTransaction)null!, limit: 1).AsTask());
+
+        // Assert
+        Assert.Equal("transaction", error.ParamName);
+    }
+
+    [Fact]
+    public async Task ShouldPropagateCancellationGivenCanceledTokenWhenQueryingPrimaryKeys()
+    {
+        // Arrange
+        var client = new InMemoryKvClient();
+        await using var transaction = await client.BeginAsync(Route, KvDurability.Async, KvMode.ReadOnly);
+        using var cancellation = new CancellationTokenSource();
+        await cancellation.CancelAsync();
+        client.ClearOperations();
+
+        // Act
+        var error = await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
+            Directory.QueryPrimaryAsync(transaction, limit: 1, ct: cancellation.Token).AsTask());
+
+        // Assert
+        Assert.Equal(cancellation.Token, error.CancellationToken);
+        Assert.Empty(client.Operations);
+    }
+
     [Fact]
     public async Task ShouldReplaceWithoutReadingGivenPreviousValue()
     {