Fix InfluxDB readback duration and superseded benchmark runs

Read the full InfluxDB corpus through one ordered native JSONL query instead of thousands of repeated sorted pages. Keep bounded row buffering, the frozen record ceiling, cancellation, full-body/digest verification and joined cleanup. Use the vendor-recommended 70% query pool within the existing physical memory budget so deduplication can execute.

Add a real million-record InfluxDB control with update, mid-stream cancellation and healthy continuation. Preserve ClickHouse paging through the common reader contract and handle the last partial native write batch. Newer Benchmarks runs cancel the previous run of their branch, including across full/contracts modes.

Validation: complete Debug solution build with warnings as errors; full formatter verification; repository governance; actionlint; real InfluxDB million-record and ClickHouse million-record controls with physical container/volume cleanup. Full session run40502233284 was cancelled at the owner request and is not complete qualification.

Author: KSemenenko

.github/workflows/benchmarks.yml

@@ -11,8 +11,8 @@ on:
         default: full
 permissions: {contents: read, actions: read}
 concurrency:
-  group: keyload-benchmarks-${{ github.ref }}${{ github.event_name == 'workflow_dispatch' && inputs.mode == 'contracts' && '-contracts' || '' }}
-  cancel-in-progress: false
+  group: keyload-benchmarks-${{ github.ref }}
+  cancel-in-progress: true
 env:
   NUGET_PACKAGES: ${{ github.workspace }}/artifacts/benchmark-runtime/packages
 jobs:
@@ -110,7 +110,7 @@ jobs:
           compression-level: 0
           retention-days: 90
       - name: Test external database bulk operations and joined cleanup
-        run: dotnet benchmarks/KeyLoad.ComparisonHost/bin/Release/net10.0/KeyLoad.ComparisonHost.dll --output Detailed --maximum-parallel-tests 1 --minimum-expected-tests 7 --treenode-filter '/*/*/NativeDatabaseTests/*' --results-directory "$GITHUB_WORKSPACE/benchmarks/KeyLoad.ComparisonHost/bin/Release/net10.0/TestResults/NativeAdapters" --report-trx --timeout 40m
+        run: dotnet benchmarks/KeyLoad.ComparisonHost/bin/Release/net10.0/KeyLoad.ComparisonHost.dll --output Detailed --maximum-parallel-tests 1 --minimum-expected-tests 8 --treenode-filter '/*/*/NativeDatabaseTests/*' --results-directory "$GITHUB_WORKSPACE/benchmarks/KeyLoad.ComparisonHost/bin/Release/net10.0/TestResults/NativeAdapters" --report-trx --timeout 40m
       - name: Test bounded terminal execution diagnostics
         run: dotnet benchmarks/KeyLoad.ComparisonHost/bin/Release/net10.0/KeyLoad.ComparisonHost.dll --output Detailed --minimum-expected-tests 1 --treenode-filter '/*/*/ComparisonHostCaptureTests/*' --results-directory "$GITHUB_WORKSPACE/benchmarks/KeyLoad.ComparisonHost/bin/Release/net10.0/TestResults/TerminalDiagnostics" --report-trx
       - name: Save original preparation diagnostics

benchmarks/KeyLoad.ComparisonHost/DatabaseAdapters/Cases/InfluxDocumentReadback.cs

@@ -0,0 +1,33 @@
+using KeyLoad.Comparisons;
+
+namespace KeyLoad.ComparisonHost.BenchmarkComparisons;
+
+internal static class InfluxDocumentReadback
+{
+    private const int Empty = 0, Clients = 16, Records = 1_000_000, Single = 1, Updated = 1;
+
+    internal static async Task VerifyAsync(IComparisonTarget target, CancellationToken token)
+    {
+        var documents = (IDocumentComparisonTarget)target;
+        var corpus = new DocumentComparisonCorpus(Empty, Clients);
+        await documents.InitializeDocumentsAsync(new(corpus, false, Records + Single), token);
+        await NativeDatabaseTests.CreateBatchesAsync(documents, corpus, Records, token);
+        await using var session = await documents.OpenDocumentSessionAsync(token);
+        await session.ExecuteAsync(Scenario.DocumentUpdate, corpus.Replacement(Updated), token);
+        await using var independent = await documents.OpenDocumentSessionAsync(token);
+        var expected = Enumerable.Range(Empty, Records)
+            .Select(index => index == Updated ? corpus.Replacement(index) : corpus.CreateDocument(index));
+        var observed = await DocumentComparisonOracle.VerifyAsync(independent.ReadDocumentsAsync(token), expected, token);
+        await Assert.That(observed.Records).IsEqualTo(Records);
+        using var cancelled = CancellationTokenSource.CreateLinkedTokenSource(token);
+        await using (var cursor = independent.ReadDocumentsAsync(cancelled.Token).GetAsyncEnumerator(cancelled.Token))
+        {
+            await Assert.That(await cursor.MoveNextAsync()).IsTrue();
+            await cancelled.CancelAsync();
+            await Assert.ThrowsAsync<OperationCanceledException>(() => cursor.MoveNextAsync().AsTask());
+        }
+        await session.CreateDocumentsAsync([corpus.CreateDocument(Records)], token);
+        await Assert.That(BenchmarkDataset.SameDocument((await independent.ReadAsync(corpus.CreateDocument(Records), token)).Value,
+            corpus.CreateDocument(Records))).IsTrue();
+    }
+}

benchmarks/KeyLoad.ComparisonHost/DatabaseAdapters/Cases/NativeDatabaseTests.cs

@@ -19,6 +19,11 @@ public Task NativeBulkPreservesBodiesAndIndependentClients(string target, int no
     public Task MongoReplicaSetReadinessPrecedesIndependentBulkClients()
         => AdditionalDatabaseResources.RunAsync(MongoDb, VerifyAsync, TestContext.Current!.Execution.CancellationToken, ReplicaNodes);
 
+    [Test]
+    public Task InfluxBulkReadbackPreservesCompleteBodiesAndHealthyContinuation()
+        => AdditionalDatabaseResources.RunAsync(BenchmarkAdditionalTargets.InfluxDb, InfluxDocumentReadback.VerifyAsync,
+            TestContext.Current!.Execution.CancellationToken);
+
     private static async Task VerifyAsync(IComparisonTarget target, CancellationToken token)
     {
         var documents = (IDocumentComparisonTarget)target;
@@ -60,7 +65,7 @@ await Assert.That(BenchmarkDataset.SameDocument((await independent.ReadAsync(cor
             corpus.CreateDocument(count))).IsTrue();
     }
 
-    private static async Task CreateBatchesAsync(IDocumentComparisonTarget target, DocumentComparisonCorpus corpus,
+    internal static async Task CreateBatchesAsync(IDocumentComparisonTarget target, DocumentComparisonCorpus corpus,
         int count, CancellationToken token)
     {
         var sessions = new List<IDocumentComparisonSession>();
@@ -79,7 +84,7 @@ private static async Task CreateBatchesAsync(IDocumentComparisonTarget target, D
                 for (var first = worker * batch; first < count; first += Clients * batch)
                 {
                     using var deadline = ComparisonDeadline.Create(BenchmarkSessionLimits.TimeoutSeconds, TimeProvider.System, token);
-                    await session.CreateDocumentsAsync([.. Enumerable.Range(first, batch).Select(corpus.CreateDocument)], deadline.Token);
+                    await session.CreateDocumentsAsync([.. Enumerable.Range(first, Math.Min(batch, count - first)).Select(corpus.CreateDocument)], deadline.Token);
                 }
             }).ToArray();
             start.SetResult();

benchmarks/KeyLoad.Comparisons/Targets/ClickHouse/Execution/ClickHouseDocumentStore.cs

@@ -99,7 +99,9 @@ internal override Task<JsonDocument> ReadAsync(string id, int? number, Cancellat
         return QueryAsync(Select + Table + Predicate + (number.HasValue ? NumericPredicate : string.Empty) + Limit + JsonFormat, parameters, token);
     }
 
-    internal override Task<JsonDocument> PageAsync(string after, CancellationToken token)
+    internal override IAsyncEnumerable<FoundDocument> ReadDocumentsAsync(CancellationToken token) => ReadPagesAsync(PageAsync, token);
+
+    private Task<JsonDocument> PageAsync(string after, CancellationToken token)
         => QueryAsync(Select + Table + Page + Policy.ReadbackBatchCapacity.ToString(CultureInfo.InvariantCulture) + JsonFormat,
             new Dictionary<string, string>(StringComparer.Ordinal) { [After] = after }, token);
 

benchmarks/KeyLoad.Comparisons/Targets/HttpDocuments/Contracts/HttpDocumentStore.cs

@@ -1,5 +1,6 @@
 using System.Buffers;
 using System.Net.Http.Headers;
+using System.Runtime.CompilerServices;
 using System.Text;
 using System.Text.Json;
 using Microsoft.Extensions.Options;
@@ -25,7 +26,7 @@ internal abstract class HttpDocumentStore(HttpClient http, string name, IOptions
     internal abstract int BatchRecords { get; }
     internal abstract Task CreateAsync(IReadOnlyList<BenchmarkDocument> documents, CancellationToken token);
     internal abstract Task<JsonDocument> ReadAsync(string id, int? number, CancellationToken token);
-    internal abstract Task<JsonDocument> PageAsync(string after, CancellationToken token);
+    internal abstract IAsyncEnumerable<FoundDocument> ReadDocumentsAsync(CancellationToken token);
     internal abstract Task IndexAsync(CancellationToken token);
 
     internal static FoundDocument[] Documents(JsonDocument document)
@@ -37,6 +38,26 @@ internal static FoundDocument[] Documents(JsonDocument document)
             row.GetProperty(Id).GetString()!, row.GetProperty(Body).GetString()!)).ToArray();
     }
 
+    protected static async IAsyncEnumerable<FoundDocument> ReadPagesAsync(Func<string, CancellationToken, Task<JsonDocument>> page,
+        [EnumeratorCancellation] CancellationToken token)
+    {
+        var after = string.Empty;
+        while (true)
+        {
+            using var result = await page(after, token).ConfigureAwait(false);
+            var rows = Documents(result);
+            if (rows.Length == 0)
+            {
+                yield break;
+            }
+            foreach (var row in rows)
+            {
+                after = row.Id;
+                yield return row;
+            }
+        }
+    }
+
     // Encodes each newline-terminated row straight to UTF-8, so a batch never holds a UTF-16 copy of its whole body.
     protected static ReadOnlyMemoryContent Rows(string head, IReadOnlyList<BenchmarkDocument> documents, Func<BenchmarkDocument, string> row)
     {

benchmarks/KeyLoad.Comparisons/Targets/HttpDocuments/Execution/HttpDocumentSession.cs

@@ -50,23 +50,15 @@ public async IAsyncEnumerable<FoundDocument> ReadDocumentsAsync([EnumeratorCance
     {
         const string InvalidOrder = "HttpDocumentDatabaseReadbackOrderRejected";
         var after = string.Empty;
-        while (true)
+        await foreach (var row in store.ReadDocumentsAsync(cancellationToken).ConfigureAwait(false))
         {
-            using var result = await store.PageAsync(after, cancellationToken).ConfigureAwait(false);
-            var rows = HttpDocumentStore.Documents(result);
-            if (rows.Length == EmptyCount)
+            cancellationToken.ThrowIfCancellationRequested();
+            if (string.CompareOrdinal(row.Id, after) <= EmptyCount)
             {
-                yield break;
-            }
-            foreach (var row in rows)
-            {
-                if (string.CompareOrdinal(row.Id, after) <= EmptyCount)
-                {
-                    throw new ComparisonFailureException(InvalidOrder);
-                }
-                after = row.Id;
-                yield return row;
+                throw new ComparisonFailureException(InvalidOrder);
             }
+            after = row.Id;
+            yield return row;
         }
     }
 

benchmarks/KeyLoad.Comparisons/Targets/InfluxDB/Execution/InfluxDocumentStore.cs

@@ -1,5 +1,6 @@
 using System.Globalization;
 using System.Net.Http.Json;
+using System.Runtime.CompilerServices;
 using System.Text;
 using System.Text.Json;
 using Microsoft.Extensions.Options;
@@ -25,8 +26,8 @@ internal sealed class InfluxDocumentStore(HttpClient http, string name, IOptions
     private const string Select = "SELECT id,body FROM documents WHERE id=$id";
     private const string NumericPredicate = " AND number>=$number AND number<$number+1";
     private const string Limit = " ORDER BY id LIMIT 2";
-    private const string Page = "SELECT id,body FROM documents WHERE id>$after ORDER BY id LIMIT ";
-    private const string Id = "id", Number = "number", After = "after", Json = "json";
+    private const string Readback = "SELECT id,body FROM documents ORDER BY id";
+    private const string Id = "id", Number = "number", Json = "json", JsonLines = "jsonl";
     private const string LinePrefix = "documents,id=";
     private const string BodyField = " body=\"", NumberField = "\",number=", IntegerSuffix = "i ";
     private const string Backslash = "\\", EscapedBackslash = "\\\\", Quote = "\"", EscapedQuote = "\\\"";
@@ -131,9 +132,15 @@ internal override Task<JsonDocument> ReadAsync(string id, int? number, Cancellat
         return QueryAsync(Select + (number.HasValue ? NumericPredicate : string.Empty) + Limit, parameters, token);
     }
 
-    internal override Task<JsonDocument> PageAsync(string after, CancellationToken token)
-        => QueryAsync(Page + Policy.ReadbackBatchCapacity.ToString(CultureInfo.InvariantCulture),
-            new Dictionary<string, object>(StringComparer.Ordinal) { [After] = after }, token);
+    internal override async IAsyncEnumerable<FoundDocument> ReadDocumentsAsync([EnumeratorCancellation] CancellationToken token)
+    {
+        using var request = new HttpRequestMessage(HttpMethod.Post, Query)
+        { Content = JsonContent.Create(new { db = Name, q = Readback, format = JsonLines }) };
+        await foreach (var row in InfluxDocumentStream.ReadAsync(Http, request, Policy, token).ConfigureAwait(false))
+        {
+            yield return row;
+        }
+    }
 
     internal override Task IndexAsync(CancellationToken token)
     {

benchmarks/KeyLoad.Comparisons/Targets/InfluxDB/Serialization/InfluxDocumentStream.cs

@@ -0,0 +1,37 @@
+using System.Runtime.CompilerServices;
+
+namespace KeyLoad.Comparisons.Targets;
+
+internal static class InfluxDocumentStream
+{
+    private const string Rejected = "InfluxReadbackRequestRejected";
+    private const string TooManyRows = "InfluxReadbackRecordLimitExceeded";
+    private const string StatusSeparator = ":";
+    private const int ErrorCharacters = 512;
+
+    internal static async IAsyncEnumerable<FoundDocument> ReadAsync(HttpClient http, HttpRequestMessage request,
+        ComparisonExecutionOptions policy, [EnumeratorCancellation] CancellationToken token)
+    {
+        using var deadline = CancellationTokenSource.CreateLinkedTokenSource(token);
+        deadline.CancelAfter(policy.OperationTimeout);
+        using var response = await http.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, deadline.Token).ConfigureAwait(false);
+        if (!response.IsSuccessStatusCode)
+        {
+            var body = await ComparisonResponse.ReadTextAsync(response.Content, policy, deadline.Token).ConfigureAwait(false);
+            var detail = new string(body.Take(ErrorCharacters).Where(character => char.IsAscii(character) && !char.IsControl(character)).ToArray());
+            throw new ComparisonFailureException(Rejected + StatusSeparator + ((int)response.StatusCode).ToString(System.Globalization.CultureInfo.InvariantCulture)
+                + StatusSeparator + detail);
+        }
+        await using var input = await response.Content.ReadAsStreamAsync(deadline.Token).ConfigureAwait(false);
+        var count = 0;
+        var maximum = checked(DocumentComparisonContract.Current.Ingestion.Records + DocumentComparisonContract.Current.Operations);
+        await foreach (var row in InfluxJsonLines.ReadAsync(input, policy, deadline).ConfigureAwait(false))
+        {
+            if (++count > maximum)
+            {
+                throw new ComparisonFailureException(TooManyRows);
+            }
+            yield return row;
+        }
+    }
+}

benchmarks/KeyLoad.Comparisons/Targets/InfluxDB/Serialization/InfluxJsonLines.cs

@@ -0,0 +1,72 @@
+using System.Buffers;
+using System.IO.Pipelines;
+using System.Text.Json;
+
+namespace KeyLoad.Comparisons.Targets;
+
+internal static class InfluxJsonLines
+{
+    private const byte Newline = (byte)'\n';
+    private const string Id = "id", Body = "body";
+    private const string TooLarge = "InfluxReadbackRowLimitExceeded", Truncated = "InfluxReadbackRowTruncated";
+
+    internal static async IAsyncEnumerable<FoundDocument> ReadAsync(Stream input, ComparisonExecutionOptions policy,
+        CancellationTokenSource deadline)
+    {
+        var reader = PipeReader.Create(input, new StreamPipeReaderOptions(bufferSize: policy.ReadBufferBytes,
+            minimumReadSize: policy.ReadBufferBytes, leaveOpen: true));
+        try
+        {
+            while (true)
+            {
+                deadline.CancelAfter(policy.OperationTimeout);
+                var result = await reader.ReadAsync(deadline.Token).ConfigureAwait(false);
+                var pending = result.Buffer;
+                while (pending.PositionOf(Newline) is { } end)
+                {
+                    deadline.Token.ThrowIfCancellationRequested();
+                    var line = pending.Slice(pending.Start, end);
+                    RequireLength(line.Length, policy.MaxResponseBytes);
+                    pending = pending.Slice(pending.GetPosition(1, end));
+                    yield return Parse(line);
+                }
+                RequireLength(pending.Length, policy.MaxResponseBytes);
+                if (result.IsCompleted)
+                {
+                    RequireComplete(pending);
+                }
+                reader.AdvanceTo(pending.Start, pending.End);
+                if (result.IsCompleted)
+                {
+                    yield break;
+                }
+            }
+        }
+        finally
+        {
+            await reader.CompleteAsync().ConfigureAwait(false);
+        }
+    }
+
+    private static FoundDocument Parse(ReadOnlySequence<byte> line)
+    {
+        using var json = JsonDocument.Parse(line);
+        return new(json.RootElement.GetProperty(Id).GetString()!, json.RootElement.GetProperty(Body).GetString()!);
+    }
+
+    private static void RequireLength(long length, int maximum)
+    {
+        if (length > maximum)
+        {
+            throw new ComparisonFailureException(TooLarge);
+        }
+    }
+
+    private static void RequireComplete(ReadOnlySequence<byte> remaining)
+    {
+        if (!remaining.IsEmpty)
+        {
+            throw new ComparisonFailureException(Truncated);
+        }
+    }
+}

build/KeyLoad.LoadTests.AppHost/Configuration/BenchmarkDeploymentOptions.cs

@@ -9,6 +9,8 @@ internal sealed record BenchmarkDeploymentOptions
     private const int DefaultNeo4jMaximumHeap = 512;
     private const int DefaultNeo4jPageCache = 256;
     private const int DefaultOpenSearchHeap = 1024;
+    private const int DefaultInfluxQueryMemoryPercent = 70;
+    private const int MaximumQueryMemoryPercent = 90;
     private const int DefaultPostgresSlotRetention = 512;
     private const int DefaultDocumentPostgresConnections = 512;
     private const int MinimumDocumentPostgresConnections = 512;
@@ -21,11 +23,13 @@ internal sealed record BenchmarkDeploymentOptions
     public int Neo4jMaximumHeapMegabytes { get; init; } = DefaultNeo4jMaximumHeap;
     public int Neo4jPageCacheMegabytes { get; init; } = DefaultNeo4jPageCache;
     public int OpenSearchHeapMegabytes { get; init; } = DefaultOpenSearchHeap;
+    public int InfluxQueryMemoryPercent { get; init; } = DefaultInfluxQueryMemoryPercent;
     public int PostgresSlotWalRetentionMegabytes { get; init; } = DefaultPostgresSlotRetention;
     public int DocumentPostgresMaximumConnections { get; init; } = DefaultDocumentPostgresConnections;
     internal bool IsValid() => DocumentPostgresMaximumConnections is >= MinimumDocumentPostgresConnections and <= MaximumDocumentPostgresConnections
         && ValidMemory(Neo4jInitialHeapMegabytes) && ValidMemory(Neo4jMaximumHeapMegabytes)
         && Neo4jInitialHeapMegabytes <= Neo4jMaximumHeapMegabytes && ValidMemory(Neo4jPageCacheMegabytes)
-        && ValidMemory(OpenSearchHeapMegabytes) && PostgresSlotWalRetentionMegabytes is >= 1 and <= MaximumWalMegabytes;
+        && ValidMemory(OpenSearchHeapMegabytes) && InfluxQueryMemoryPercent is >= 1 and <= MaximumQueryMemoryPercent
+        && PostgresSlotWalRetentionMegabytes is >= 1 and <= MaximumWalMegabytes;
     private static bool ValidMemory(int value) => value is >= 1 and <= MaximumMemoryMegabytes;
 }