diff --git a/CHANGELOG.md b/CHANGELOG.md index 6dbc92b..e566e53 100644 --- a/CHANGELOG.md +++ b/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.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 diff --git a/Directory.Build.props b/Directory.Build.props index b3d7c3a..d52ec64 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -11,7 +11,7 @@ - 1.4.0 + 1.4.1 $(PackageVersion) 1.0.0.0 $(PackageVersion) @@ -26,7 +26,7 @@ portable README.md fitz distributed-systems kv queue rpc lease notice stream schedule - Adds transaction route reporting and caller-owned transaction queries for Cntryl.Fitz.Extensions. See CHANGELOG.md. + Adds bounded read-only primary-key paging for Cntryl.Fitz.Extensions directories. See CHANGELOG.md. $(MSBuildThisFileDirectory)artifacts\packages diff --git a/README.md b/README.md index 336c709..1bc8603 100644 --- a/README.md +++ b/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( @@ -144,6 +144,9 @@ 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)` @@ -151,6 +154,10 @@ read forms are shorthand that open a short read-only transaction. Cursors are bo 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 diff --git a/src/Fitz.Extensions/KvDirectory.cs b/src/Fitz.Extensions/KvDirectory.cs index ee0934d..23df674 100644 --- a/src/Fitz.Extensions/KvDirectory.cs +++ b/src/Fitz.Extensions/KvDirectory.cs @@ -194,6 +194,55 @@ public async ValueTask> QueryAsync( return await ExecuteAsync(transaction, plan, ct).ConfigureAwait(false); } + /// + /// 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 + /// and this directory's primary rows, so a cursor issued for another route or query is rejected. + /// + /// The caller-owned transaction used for the scan. + /// The maximum number of primary records to return. + /// An opaque continuation cursor from a previous primary page. + /// Cancellation token. + /// Primary records in ascending key order and a cursor for the next page, if any. + /// The transaction does not report its route. + public async ValueTask> 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); + } + + /// + /// 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. + /// + /// The KV client used to open the read-only transaction. + /// The exact KV route containing the directory. + /// The maximum number of primary records to return. + /// An opaque continuation cursor from a previous primary page. + /// Cancellation token. + /// Primary records in ascending key order and a cursor for the next page, if any. + [SuppressMessage("Reliability", "CA2007:Consider calling ConfigureAwait on the awaited task", + Justification = "The await-using declaration must retain the transaction type.")] + public async ValueTask> 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); + } + /// /// 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 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> ExecuteAsync(IKvTransaction transaction, QueryPlan plan, CancellationToken ct) { var matches = new List(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 key) diff --git a/src/Fitz.Extensions/Page.cs b/src/Fitz.Extensions/Page.cs index 6a64bb3..24f7047 100644 --- a/src/Fitz.Extensions/Page.cs +++ b/src/Fitz.Extensions/Page.cs @@ -2,7 +2,7 @@ namespace Cntryl.Fitz.Extensions; /// /// One keyset-ordered directory page. 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. /// /// The item type. /// The items in this page, in the query's effective order. diff --git a/test/Fitz.Extensions.Tests/KvDirectoryTests.cs b/test/Fitz.Extensions.Tests/KvDirectoryTests.cs index 7b43274..4aa3359 100644 --- a/test/Fitz.Extensions.Tests/KvDirectoryTests.cs +++ b/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(() => + Directory.QueryPrimaryAsync(other, limit: 1, cursor: first.NextCursor).AsTask()); + var queryError = await Assert.ThrowsAsync(() => 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(() => + 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(() => + 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(() => 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(() => + 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(() => + 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() {