Fix Batching Coordinator Task Explosion (#8832)

Author: michaelstaib

.clabot

@@ -0,0 +1,7 @@
+{
+  "contributors": [],
+  "allowlist": [
+    "github-actions[bot]",
+    "dependabot[bot]"
+  ]
+}

.github/workflows/ci.yml

@@ -5,6 +5,8 @@ on:
     branches:
       - main
       - main-version-*
+    paths-ignore:
+      - 'src/HotChocolate/AspNetCore/benchmarks/k6/performance-data.json'
 
 concurrency:
   group: ci-new-2-${{ github.event.pull_request.number }}
@@ -257,9 +259,170 @@ jobs:
           flags: unittests
           fail_ci_if_error: true
 
+  performance-tests:
+    name: "Performance Tests"
+    needs: check-changes
+    if: needs.check-changes.outputs.src_changes == 'true' && github.event.pull_request.draft == false
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      pull-requests: write
+
+    steps:
+      - name: Checkout PR code
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          show-progress: false
+
+      - name: Setup .NET
+        uses: actions/setup-dotnet@v4
+        with:
+          dotnet-version: |
+            8.x
+            9.x
+            10.x
+
+      - name: Install k6
+        run: |
+          sudo gpg -k
+          sudo gpg --no-default-keyring --keyring /usr/share/keyrings/k6-archive-keyring.gpg --keyserver hkp://keyserver.ubuntu.com:80 --recv-keys C5AD17C747E3415A3642D57D77C6C491D6AC1D69
+          echo "deb [signed-by=/usr/share/keyrings/k6-archive-keyring.gpg] https://dl.k6.io/deb stable main" | sudo tee /etc/apt/sources.list.d/k6.list
+          sudo apt-get update
+          sudo apt-get install k6
+
+      - name: Install jq
+        run: sudo apt-get install -y jq
+
+      - name: Restore dependencies
+        run: dotnet restore src/HotChocolate/AspNetCore/benchmarks/k6/eShop.slnx
+
+      - name: Start AppHost and wait for readiness
+        working-directory: src/HotChocolate/AspNetCore/benchmarks/k6/Catalog.AppHost
+        run: |
+          echo "Starting AppHost..."
+          dotnet run > /tmp/apphost.log 2>&1 &
+          APPHOST_PID=$!
+          echo "APPHOST_PID=$APPHOST_PID" >> $GITHUB_ENV
+
+          echo "Waiting for server to be ready..."
+          for i in {1..30}; do
+            if curl -s -o /dev/null -w "%{http_code}" http://localhost:5224/graphql -X POST \
+              -H "Content-Type: application/json" \
+              -d '{"query": "{ __typename }"}' | grep -q "200"; then
+              echo "Server is ready!"
+              break
+            fi
+            echo "Waiting... ($i/30)"
+            sleep 2
+          done
+
+      - name: Run performance tests and collect data
+        working-directory: src/HotChocolate/AspNetCore/benchmarks/k6
+        run: |
+          chmod +x run-and-collect.sh
+          ./run-and-collect.sh performance-data-current.json
+
+      - name: Stop AppHost
+        if: always()
+        run: |
+          if [ -n "$APPHOST_PID" ]; then
+            kill $APPHOST_PID 2>/dev/null || true
+            wait $APPHOST_PID 2>/dev/null || true
+          fi
+
+      - name: Commit and push performance data to current branch
+        working-directory: src/HotChocolate/AspNetCore/benchmarks/k6
+        run: |
+          # Copy the performance data to the tracked filename
+          cp performance-data-current.json performance-data.json
+
+          # Configure git
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+
+          # Add and commit the performance data
+          git add performance-data.json
+
+          # Only commit if there are changes
+          if ! git diff --staged --quiet; then
+            git commit -m "Update performance data [skip ci]"
+            git push origin HEAD:${{ github.head_ref }}
+          else
+            echo "No changes to performance data"
+          fi
+
+      - name: Fetch baseline performance data from main
+        run: |
+          git fetch origin main:main
+          if git show main:src/HotChocolate/AspNetCore/benchmarks/k6/performance-data.json > baseline-performance.json 2>/dev/null; then
+            echo "Baseline data fetched successfully"
+          else
+            echo "No baseline data found on main branch"
+            echo '{}' > baseline-performance.json
+          fi
+
+      - name: Compare performance and generate report
+        working-directory: src/HotChocolate/AspNetCore/benchmarks/k6
+        run: |
+          chmod +x compare-performance.sh
+          ./compare-performance.sh performance-data-current.json ../../../../baseline-performance.json performance-report.md
+
+      - name: Comment PR with performance report
+        uses: actions/github-script@v7
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          script: |
+            const fs = require('fs');
+            const reportPath = 'src/HotChocolate/AspNetCore/benchmarks/k6/performance-report.md';
+
+            let report;
+            try {
+              report = fs.readFileSync(reportPath, 'utf8');
+            } catch (error) {
+              console.error('Failed to read performance report:', error);
+              return;
+            }
+
+            // Add timestamp and commit info to the report
+            const timestamp = new Date().toUTCString();
+            const commitSha = context.sha.substring(0, 7);
+            const runNumber = context.runNumber;
+
+            const commentBody = `${report}\n\n---\n*Run #${runNumber} • Commit ${commitSha} • ${timestamp}*`;
+
+            // Always create a new comment
+            await github.rest.issues.createComment({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              issue_number: context.issue.number,
+              body: commentBody,
+            });
+
+      - name: Upload performance data as artifact
+        uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: performance-data
+          path: |
+            src/HotChocolate/AspNetCore/benchmarks/k6/performance-data-current.json
+            src/HotChocolate/AspNetCore/benchmarks/k6/performance-report.md
+            /tmp/apphost.log
+          retention-days: 30
+
+      - name: Check for performance regression
+        working-directory: src/HotChocolate/AspNetCore/benchmarks/k6
+        run: |
+          # Fail the build if there's a significant performance regression
+          if grep -q "⚠️ \*\*Performance regression detected" performance-report.md; then
+            echo "::warning::Performance regression detected! Please review the performance report."
+            # Uncomment the next line to fail the build on regression
+            # exit 1
+          fi
+
   ci-status-check:
     name: "CI Status Check"
-    needs: [library-tests, website-tests]
+    needs: [library-tests, website-tests, performance-tests]
     if: always()
     runs-on: ubuntu-latest
     steps:
@@ -268,4 +431,5 @@ jobs:
         if: |
           always() &&
           (needs.library-tests.result == 'failure' ||
-           needs.website-tests.result == 'failure')
+           needs.website-tests.result == 'failure' ||
+           needs.performance-tests.result == 'failure')

src/All.slnx

@@ -35,6 +35,12 @@
     <Project Path="HotChocolate/ApolloFederation/test/ApolloFederation.Tests/HotChocolate.ApolloFederation.Tests.csproj" />
   </Folder>
   <Folder Name="/HotChocolate/AspNetCore/" />
+  <Folder Name="/HotChocolate/AspNetCore/benchmarks/" />
+  <Folder Name="/HotChocolate/AspNetCore/benchmarks/k6/">
+    <Project Path="HotChocolate/AspNetCore/benchmarks/k6/Catalog.AppHost/eShop.Catalog.AppHost.csproj" />
+    <Project Path="HotChocolate/AspNetCore/benchmarks/k6/Catalog.API/eShop.Catalog.API.csproj" />
+    <Project Path="HotChocolate/AspNetCore/benchmarks/k6/Catalog.ServiceDefaults/eShop.Catalog.ServiceDefaults.csproj" />
+  </Folder>
   <Folder Name="/HotChocolate/AspNetCore/src/">
     <Project Path="HotChocolate/AspNetCore/src/AspNetCore.Authorization.Opa/HotChocolate.AspNetCore.Authorization.Opa.csproj" />
     <Project Path="HotChocolate/AspNetCore/src/AspNetCore.Authorization/HotChocolate.AspNetCore.Authorization.csproj" />

src/Directory.Packages.props

@@ -9,6 +9,7 @@
     <PackageVersion Include="Aspire.Hosting.RabbitMQ" Version="9.4.1" />
     <PackageVersion Include="Aspire.Hosting.Redis" Version="9.4.1" />
     <PackageVersion Include="Aspire.Hosting" Version="9.4.1" />
+    <PackageVersion Include="Aspire.AppHost.Sdk" Version="1.9.0" />
     <PackageVersion Include="Azure.Storage.Blobs" Version="12.23.0" />
     <PackageVersion Include="Basic.Reference.Assemblies.Net100" Version="1.8.3" />
     <PackageVersion Include="Basic.Reference.Assemblies.Net80" Version="1.8.3" />

src/GreenDonut/src/GreenDonut.Abstractions/Batch.cs

@@ -38,6 +38,12 @@ public abstract class Batch
     /// </summary>
     public abstract BatchStatus Status { get; }
 
+    /// <summary>
+    /// Gets a high-resolution timestamp representing when this batch was first created.
+    /// This value is used to enforce maximum batch age timeouts to prevent starvation.
+    /// </summary>
+    public abstract long CreatedTimestamp { get; }
+
     /// <summary>
     /// Gets a high-resolution timestamp from representing the last time an item was added to this batch.
     /// This value is used for recency checks in scheduling decisions.

src/GreenDonut/src/GreenDonut/Batch.cs

@@ -12,7 +12,9 @@ internal sealed class Batch<TKey> : Batch where TKey : notnull
     private Func<Batch<TKey>, CancellationToken, ValueTask> _dispatch = null!;
     private CancellationToken _ct = CancellationToken.None;
     private int _status = Enqueued;
-    private long _timestamp;
+    private long _createdTimestamp;
+    private long _modifiedTimestamp;
+    private CancellationTokenRegistration _promiseCancellationRegistration;
 
     public bool IsScheduled { get; set; }
 
@@ -22,7 +24,9 @@ internal sealed class Batch<TKey> : Batch where TKey : notnull
 
     public override BatchStatus Status => (BatchStatus)_status;
 
-    public override long ModifiedTimestamp => _timestamp;
+    public override long CreatedTimestamp => _createdTimestamp;
+
+    public override long ModifiedTimestamp => _modifiedTimestamp;
 
     public override bool Touch()
     {
@@ -36,7 +40,7 @@ public Promise<TValue> GetOrCreatePromise<TValue>(TKey key, bool allowCachePropa
         // as long as there are components interacting with this batch its good to
         // keep it in enqueued state.
         Interlocked.Exchange(ref _status, Enqueued);
-        _timestamp = Stopwatch.GetTimestamp();
+        _modifiedTimestamp = Stopwatch.GetTimestamp();
 
         if (_items.TryGetValue(key, out var value))
         {
@@ -55,13 +59,19 @@ public Promise<TValue> GetPromise<TValue>(TKey key)
         => (Promise<TValue>)_items[key];
 
     public override async Task DispatchAsync()
-        => await _dispatch(this, _ct);
+    {
+        await _dispatch(this, _ct);
+        _promiseCancellationRegistration.Dispose();
+    }
 
     internal void Initialize(Func<Batch<TKey>, CancellationToken, ValueTask> dispatch, CancellationToken ct)
     {
         _status = Enqueued;
         _dispatch = dispatch;
         _ct = ct;
+        _createdTimestamp = Stopwatch.GetTimestamp();
+        _modifiedTimestamp = _createdTimestamp;
+        _promiseCancellationRegistration = ct.Register(TryCancelPromises);
     }
 
     internal void ClearUnsafe()
@@ -72,5 +82,16 @@ internal void ClearUnsafe()
         _status = Enqueued;
         _dispatch = null!;
         _ct = CancellationToken.None;
+        _createdTimestamp = 0;
+        _modifiedTimestamp = 0;
+        _promiseCancellationRegistration.Dispose();
+    }
+
+    internal void TryCancelPromises()
+    {
+        foreach (var item in _items.Values)
+        {
+            item.TryCancel();
+        }
     }
 }

src/GreenDonut/src/GreenDonut/Instrumentation/AggregateDataLoaderDiagnosticEventListener.cs

@@ -60,6 +60,46 @@ public override void BatchItemError<TKey>(
         }
     }
 
+    /// <inheritdoc />
+    public override IDisposable RunBatchDispatchCoordinator()
+    {
+        var scopes = new IDisposable[listeners.Length];
+
+        for (var i = 0; i < listeners.Length; i++)
+        {
+            scopes[i] = listeners[i].RunBatchDispatchCoordinator();
+        }
+
+        return new AggregateEventScope(scopes);
+    }
+
+    /// <inheritdoc />
+    public override void BatchDispatchError(Exception error)
+    {
+        for (var i = 0; i < listeners.Length; i++)
+        {
+            listeners[i].BatchDispatchError(error);
+        }
+    }
+
+    /// <inheritdoc />
+    public override void BatchEvaluated(int openBatches)
+    {
+        for (var i = 0; i < listeners.Length; i++)
+        {
+            listeners[i].BatchEvaluated(openBatches);
+        }
+    }
+
+    /// <inheritdoc />
+    public override void BatchDispatched(int dispatchedBatches, bool inParallel)
+    {
+        for (var i = 0; i < listeners.Length; i++)
+        {
+            listeners[i].BatchDispatched(dispatchedBatches, inParallel);
+        }
+    }
+
     private sealed class AggregateEventScope(IDisposable[] scopes) : IDisposable
     {
         public void Dispose()

src/GreenDonut/src/GreenDonut/Instrumentation/DataLoaderDiagnosticEventListener.cs

@@ -46,6 +46,22 @@ public virtual void BatchItemError<TKey>(
         where TKey : notnull
     { }
 
+    /// <inheritdoc />
+    public virtual IDisposable RunBatchDispatchCoordinator()
+        => EmptyScope;
+
+    /// <inheritdoc />
+    public virtual void BatchDispatchError(Exception error)
+    { }
+
+    /// <inheritdoc />
+    public virtual void BatchEvaluated(int openBatches)
+    { }
+
+    /// <inheritdoc />
+    public virtual void BatchDispatched(int dispatchedBatches, bool inParallel)
+    { }
+
     private sealed class EmptyActivityScope : IDisposable
     {
         public void Dispose() { }