bytefish.de

Scaling Autonomous AI Agents: Multi-Tenancy with SqlFlow, CitusDB, NATS and .NET Aspire

SqlFlow is a simple durable execution workflow system for PostgreSQL and SQL Server. It handles scheduling, state checkpointing, and retries without needing heavy external workflow engines like Temporal or Cadence:

In this article we'll take a look at how to horizontally scale our system.

Table of contents

Whats the Problem?

In my last article I have shown how to use SqlFlow for building resilient, durable workflows running on a single Postgres instance. That's totally sufficient for 90% of the use cases, because modern servers take you a long, long way and vertically scale well.

But what if we have thousands of customers and a single Postgres instance is unable to handle concurrent access to the database anymore? Or the storage runs out? What if it's locking too many rows? At some point we'll need to scale horizontally and as soon as you are going distributed... systems get complex.

I wanted to prepared for scaling, so I've added a tenant identifier to the SqlFlow framework. It can be used as a stable partition key for sharding our data. In this article, we'll take a look at distributing and sharding our data using CitusDB and scaling with NATS.

Solving Signaling Scale: NATS JetStream

The biggest issue when using a database as a queue is constantly polling the database to check for work.

Polling works fine, if you have single digit workers and long polling intervals. But imagine you've got 100 worker nodes, that constantly hammer the database in 1 second intervals... without any new work being available.

So I've introduced the concept of Signals into the SqlFlow framework, effectively separating the system into a "control plane" and a "storage plane". It allows us to use database native features, such as LISTEN and NOTIFY primitives, instead of polling.

But using LISTEN/NOTIFY or SQL Server Service Brokers introduce the thundering herd problem.

Info

The thundering herd problem occurs when a single spawned task triggers a broadcast notification, that wakes up every connected worker simultaneously. This usually causes them all to query the database and burn through CPU cycles, even though only one worker can actually claim the task.

That's why I am also providing signaling using NATS. NATS JetStream has the concept of so called Consumer Groups , which guarantees, that a single message is delivered to exactely one worker in the group. By using this, we basically end up with 1-1 ratio between wake-up messages and database queries.

Solving Storage Scale: Horizontal Sharding with CitusDB

CitusDB is a PostgreSQL extension and it extends PostgreSQL with distributed sharding. Instead of writing all data to a single node, the CitusDB extension tries to distribute the load horizontally on all machines in the cluster.

Warning

Before we start a word of caution:

Everything in software development is a trade-off, so is using CitusDB. While it solves the problem of distributing your data nicely, it introduces a high operational complexity. My advice is to start on a single machine and use a tenant_id.

Only introduce CitusDB when your metrics prove that your single database node's CPU, disk IOPS or connection limits have hit a wall.

Solving Developer Experience: Local Development with .NET Aspire

Orchestrating and Debugging distributed systems is hard. We've got a CitusDB Coordinator, 10 CitusDB shards, we need to deploy the database schema, we need to start the application, that uses the SqlFlow SDK to create queues.

I am tired of writing docker-compose files, and there's a much better way: Aspire.

What we are going to build

We are going to horizontally scale out our SqlFlow "AI Workflow Example", that bascially models how an agent interacts with a user and uses SqlFlow primitives to build a resilient agent workflow.

Solution Overview

The solution consists of three .NET projects:

  1. SqlFlowSdk.DistributedSample.AppHost (Orchestration Layer)

Serves as the central entry point and service orchestrator using .NET Aspire.

  • CitusDB Cluster: Spins up 11 PostgreSQL/Citus containers. One Coordinator node (coordinator, exposed on port 5432) and ten Worker nodes (worker-1 through worker-10) communicating over Aspire's internal Docker network.
  • NATS JetStream Broker: Runs an isolated nats:2.10 container with JetStream enabled (-js) on port 4222 to handle real-time wake-up signals and human-in-the-loop approval events.

  • Lifecycle Management: Enforces strict startup ordering using WaitFor and WaitForCompletion so that the database schema and shards are fully initialized before any workers begin processing traffic.

  • SqlFlowSdk.DistributedSample.DbDeployer (Database Initialization & Sharding)

A short-lived console application that runs once at startup after the Citus coordinator and all 10 worker containers are ready.

  • Schema Provisioning: Executes ssf-postgres-minimal.sql to create the ssf schema, core tables (tasks, runs, checkpoints), and stored procedures.

  • Cluster Registration and Sharding: Registers the coordinator and all 10 worker nodes (citus_add_node), then distributes the core tables across the cluster using tenant_id as the distribution column (create_distributed_table). This guarantees tenant isolation and co-located single-shard execution.

  • SqlFlowSdk.DistributedSample.WorkerApp (API & Background Agent Execution)

An ASP.NET Core Web Application hosting both the HTTP ingress endpoints and the background SqlFlowSdk worker engine.

  • Webhook API: Exposes POST /agent/{tenantId}/start to dispatch new autonomous bug-fixing jobs and POST /agent/{tenantId}/review/{issueId}/{correlationId} to resume paused workflows with human feedback.
  • Hybrid Persistence and Signaling: Uses PostgreSQL/Citus as the durable source of truth (leveraging FOR UPDATE SKIP LOCKED and transactional checkpointing) while subscribing to NATS JetStream for instant task wake-ups and external approval events, falling back to smart polling if the broker is temporarily unreachable.

SqlFlowSdk.DistributedSample.AppHost

This .NET Aspire AppHost configuration orchestrates the entire distributed environment in code, spinning up a NATS JetStream broker alongside an 11-node CitusDB cluster, consisting of one coordinator and ten worker containers.

It enforces a strict startup sequence. Once the database nodes are running, the db-deployer project initializes the schema and registers the cluster shards using injected host environment variables, and only after that migration completes does the sqlflow-workers application launch on port 5000 to begin processing requests.

var builder = DistributedApplication.CreateBuilder(args);

string postgresConnectionString = "Host=127.0.0.1;Port=5432;Database=sqlflow;Username=postgres;Password=SuperStrongPassword123;";
string natsConnectionString = "nats://127.0.0.1:4222";

// NATS JetStream
var nats = builder.AddContainer("nats-broker", "nats", "2.10")
    .WithArgs("-js")
    .WithEndpoint(port: 4222, targetPort: 4222, name: "client", isProxied: false);

// CitusDB Coordinator Node
var coordinator = builder.AddContainer("coordinator", "citusdata/citus", "12.1")
    .WithEnvironment("POSTGRES_USER", "postgres")
    .WithEnvironment("POSTGRES_PASSWORD", "SuperStrongPassword123")
    .WithEnvironment("POSTGRES_DB", "sqlflow")
    .WithEnvironment("POSTGRES_HOST_AUTH_METHOD", "trust")
    .WithEndpoint(port: 5432, targetPort: 5432, name: "tcp");

// Database Deployer (declared upfront so workers can be attached directly)
var dbDeployer = builder.AddProject<Projects.SqlFlowSdk_DistributedSample_DbDeployer>("db-deployer")
    .WithEnvironment("ConnectionStrings__SqlFlowDb", postgresConnectionString)
    .WithEnvironment("CITUS_COORDINATOR_HOST", coordinator.Resource.Name)
    .WaitFor(coordinator);

// CitusDB Worker Nodes (10 Replicas) & direct attachment to the deployer
for (int i = 1; i <= 10; i++)
{
    var worker = builder.AddContainer($"worker-{i}", "citusdata/citus", "12.1")
        .WithContainerName($"sqlflow-worker-{i}")
        .WithEnvironment("POSTGRES_USER", "postgres")
        .WithEnvironment("POSTGRES_PASSWORD", "SuperStrongPassword123")
        .WithEnvironment("POSTGRES_DB", "sqlflow")
        .WithEnvironment("POSTGRES_HOST_AUTH_METHOD", "trust")
        .WithEndpoint(targetPort: 5432, name: "tcp")
        .WaitFor(coordinator);

    dbDeployer
        .WithEnvironment($"CITUS_WORKER_{i}_HOST", worker.Resource.Name)
        .WaitFor(worker);

    dbDeployer.WaitFor(worker);
}

// Worker App (waits for NATS and full completion of the deployer)
builder.AddProject<Projects.SqlFlowSdk_DistributedSample_WorkerApp>("sqlflow-workers")
    .WithEnvironment("ConnectionStrings__SqlFlowDb", postgresConnectionString)
    .WithEnvironment("ConnectionStrings__nats-broker", natsConnectionString)
    .WithHttpEndpoint(port: 5000, name: "http", isProxied: false)
    .WaitFor(nats)
    .WaitForCompletion(dbDeployer);

builder.Build().Run();

SqlFlowSdk.DistributedSample.DbDeployer

We start by creating a file called citus-db-sharding.sql to configure the CitusDB sharding. We are going to use the tenant_id as the sharding key:

-- Coordinator Host and Port Configuration using the Aspire configuration
SELECT citus_set_coordinator_host('coordinator', 5432);

-- Register the worker nodes with the coordinator using the resource names of the worker containers
DO $$
DECLARE
    i INT;
    worker_host TEXT;
BEGIN
    FOR i IN 1..10 LOOP
        worker_host := 'worker-' || i;
        IF NOT EXISTS (SELECT 1 FROM pg_dist_node WHERE nodename = worker_host) THEN
            PERFORM citus_add_node(worker_host, 5432);
        END IF;
    END LOOP;
END $$;

-- Distribute the tables based on tenant_id across the shards
DO $$
BEGIN
    -- Distribute 'ssf.tasks'
    IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'ssf' AND table_name = 'tasks')
       AND NOT EXISTS (SELECT 1 FROM pg_dist_partition WHERE logicalrelid = 'ssf.tasks'::regclass) THEN
        PERFORM create_distributed_table('ssf.tasks', 'tenant_id');
    END IF;

    -- Distribute 'ssf.runs'
    IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'ssf' AND table_name = 'runs')
       AND NOT EXISTS (SELECT 1 FROM pg_dist_partition WHERE logicalrelid = 'ssf.runs'::regclass) THEN
        PERFORM create_distributed_table('ssf.runs', 'tenant_id');
    END IF;

    -- Distribute 'ssf.checkpoints'
    IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'ssf' AND table_name = 'checkpoints')
       AND NOT EXISTS (SELECT 1 FROM pg_dist_partition WHERE logicalrelid = 'ssf.checkpoints'::regclass) THEN
        PERFORM create_distributed_table('ssf.checkpoints', 'tenant_id');
    END IF;
END $$;

And a small application is used to create the SqlFlow database and configure the distributed tables:

using Npgsql;

Console.WriteLine("=== SqlFlow DbDeployer started ===");

string connectionString = Environment.GetEnvironmentVariable("ConnectionStrings__SqlFlowDb")
    ?? "Host=127.0.0.1;Port=5432;Database=sqlflow;Username=postgres;Password=SuperStrongPassword123;";

await WaitForDatabaseAsync(connectionString);

string baseSqlPath = Path.Combine(AppContext.BaseDirectory, "sql", "ssf-postgres-minimal.sql");
string citusSqlPath = Path.Combine(AppContext.BaseDirectory, "sql", "citus-db-sharding.sql");

string baseSql = await File.ReadAllTextAsync(baseSqlPath);
string citusSql = await File.ReadAllTextAsync(citusSqlPath);

// Create Schema, Tables, Functions, etc. 
await ExecuteSqlAsync(connectionString, "SqlFlow-Schema (ssf-postgres-minimal.sql)", baseSql);

// CitusDB Sharding: Create Distributed Tables, Shards, etc.
await ExecuteCitusShardingAsync(connectionString, "CitusDB Sharding (citus-db-sharding.sql)", citusSql);

// Verify that the deployment was successful and that the schema and workers are correctly set up
await VerifyDeploymentAsync(connectionString);

Console.WriteLine("=== SqlFlow DbDeployer completed successfully ===");
return 0;

static async Task WaitForDatabaseAsync(string connStr)
{
    Console.WriteLine("Waiting for Citus and Postgres Database...");
    int maxRetries = 30;

    for (int i = 1; i <= maxRetries; i++)
    {
        try
        {
            await using var conn = new NpgsqlConnection(connStr);
            await conn.OpenAsync();
            Console.WriteLine("--> Database connection established");
            return;
        }
        catch (Exception)
        {
            if (i == maxRetries) throw;
            Console.WriteLine($"[Attempt {i}/{maxRetries}] Database is not ready yet. Retrying in 2 seconds...");
            await Task.Delay(2000);
        }
    }
}

static async Task ExecuteSqlAsync(string connStr, string description, string sql)
{
    Console.WriteLine($"Executing: {description}...");

    await using var conn = new NpgsqlConnection(connStr);
    await conn.OpenAsync();

    await using var cmd = conn.CreateCommand();
    cmd.CommandText = sql;
    await cmd.ExecuteNonQueryAsync();

    Console.WriteLine($"--> {description} executed successfully");
}

static async Task ExecuteCitusShardingAsync(string connStr, string description, string sql)
{
    Console.WriteLine($"Executing: {description} (statement by statement)...");

    await using var conn = new NpgsqlConnection(connStr);
    await conn.OpenAsync();

    // Trennt DO $$ ... END $$; Blöcke und einzelne SELECTs sauber auf, 
    // damit Npgsql sie nicht in eine gemeinsame Transaktion verpackt.
    var statements = SplitSqlStatements(sql);

    foreach (var stmt in statements)
    {
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = stmt;
        await cmd.ExecuteNonQueryAsync();
    }

    Console.WriteLine($"--> {description} executed successfully");
}

static async Task VerifyDeploymentAsync(string connStr)
{
    await using var conn = new NpgsqlConnection(connStr);
    await conn.OpenAsync();

    await using var schemaCmd = conn.CreateCommand();
    schemaCmd.CommandText = "SELECT schema_name FROM information_schema.schemata WHERE schema_name = 'ssf';";
    var schema = await schemaCmd.ExecuteScalarAsync();

    if (schema is null)
        throw new InvalidOperationException("Verification failed: Schema 'ssf' was not found after deployment!");

    await using var workerCmd = conn.CreateCommand();
    workerCmd.CommandText = "SELECT count(*) FROM citus_get_active_worker_nodes();";
    var workerCount = Convert.ToInt32(await workerCmd.ExecuteScalarAsync());

    Console.WriteLine($"--> Verified: Schema '{schema}' exists, active Citus workers: {workerCount}");
}

static IEnumerable<string> SplitSqlStatements(string sql)
{
    // Zerlegt das Skript anhand von "END $$;" oder ";" außerhalb von Dollar-Quotes
    var results = new List<string>();
    var current = new System.Text.StringBuilder();
    bool inDollarQuote = false;

    var lines = sql.Split('\n');
    foreach (var rawLine in lines)
    {
        var line = rawLine.Trim();
        if (string.IsNullOrWhiteSpace(line) || line.StartsWith("--"))
            continue;

        current.AppendLine(rawLine);

        // Zählt Vorkommen von $$ in der Zeile, um DO $$ ... END $$; Blöcke zusammenzuhalten
        int dollarCount = (line.Length - line.Replace("$$", "").Length) / 2;
        if (dollarCount % 2 != 0)
            inDollarQuote = !inDollarQuote;

        if (!inDollarQuote && line.EndsWith(";"))
        {
            results.Add(current.ToString().Trim());
            current.Clear();
        }
    }

    if (current.Length > 0)
        results.Add(current.ToString().Trim());

    return results;
}

The Worker Application

The WorkerApp is the core engine of the system. It is a single ASP.NET Core web application that handles incoming web requests and runs the AI agents in the background at the same time.

Instead of setting up separate services for the API, message queues, and background workers, WorkerApp combines them using SqlFlowSdk:

  • Starting the Agent (POST /agent/{tenantId}/start): When a request arrives to fix a bug for a specific customer (tenantId), the app saves the task to PostgreSQL. Citus automatically routes that data to the database shard assigned to that customer, and NATS sends a quick "wake up" ping so the worker starts immediately.
  • Running & Pausing for Approval: The background worker picks up the task, uses an AI service (ILlmService) to write a code fix, and saves its progress. Before creating a pull request, the agent safely pauses without blocking any server threads and waits for a human to review its work.
  • Handling Feedback (POST /agent/{tenantId}/review/...): When a human reviewer approves or rejects the fix via the webhook, NATS immediately wakes the paused agent back up. If rejected, the agent reads the reviewer's feedback and tries again (up to 3 times). Once approved, it opens the pull request on GitHub and marks the job as finished.

Because every database query is tied to a tenantId, you can easily run multiple copies of WorkerApp side by side. PostgreSQL ensures two workers never grab the same job twice, while Citus keeps each customer's data separated on its own database shard.

The AI Agent Job using the SqlFlow SDK

We define a durable, multi-tenant AI workflow that fetches a GitHub issue, drafts a code fix via an LLM and suspends execution (ctx.AwaitEvent) until a human approves or rejects it. It retries the task with feedback up to three times before opening a pull request or escalating to a senior developer.

using SqlFlowSdk.Core;
using SqlFlowSdk.DistributedSample.WorkerApp.Models;
using SqlFlowSdk.DistributedSample.WorkerApp.Services;
using System.Text.Json.Nodes;

public class AutonomousAgentJob : IJob<AgentTask, AgentResult>
{
    private readonly ILogger<AutonomousAgentJob> _logger;

    private readonly ILlmService _llmService;
    private readonly IGitHubService _gitHubService;
    private readonly ILocalNotificationService _localNotificationService;

    public AutonomousAgentJob(ILogger<AutonomousAgentJob> logger, ILlmService llmService, IGitHubService gitHubService, ILocalNotificationService localNotificationService)
    {
        _logger = logger;
        _llmService = llmService;
        _gitHubService = gitHubService;
        _localNotificationService = localNotificationService;
    }

    public async Task<AgentResult> ExecuteAsync(TaskContext ctx, AgentTask task)
    {
        _logger.LogInformation("Agent starts researching ticket {IssueId} for tenant {TenantId}", task.IssueId, task.TenantId);

        // Load the Issue Context first, so the LLM has all relevant information
        var bugReport = await ctx.Step("fetch-issue-context", async () =>
            await _gitHubService.GetIssueDetailsAsync(task.TenantId, task.IssueId, ctx.CancellationToken));

        bool isApproved = false;
        int attempt = 0;

        string lastFeedback = "Initial Attempt";

        while (!isApproved && attempt < 3)
        {
            attempt++;

            // Generate CorrelationID for the Event
            string correlationId = $"{ctx.TaskId}-attempt-{attempt}";

            _logger.LogInformation("Attempt {attempt}/3: Generating a fix for tenant {TenantId} based on: {feedback}", attempt, task.TenantId, lastFeedback);

            Solution proposedFix = await ctx.Step($"generate-code-fix-{attempt}", async () =>
                await _llmService.GenerateFixAsync(task.TenantId, bugReport.StackTrace, lastFeedback, ctx.CancellationToken));

            await ctx.Step($"notify-reviewer-{attempt}", async () =>
            {
                // Notify the reviewer via GitHub service
                await _gitHubService.RequestHumanReviewAsync(task.TenantId, task.IssueId, proposedFix, correlationId, ctx.CancellationToken);
                // Notify the reviewer via local notification service
                await _localNotificationService.NotifyReviewerAsync(task.TenantId, task.IssueId, correlationId, ctx.CancellationToken);
            });

            _logger.LogInformation("Review for tenant {TenantId}, CorrelationId {CorrelationId} has been requested. Agent goes idle and waits...", task.TenantId, correlationId);

            // Wait for a human decision without blocking a thread
            JsonNode? review = await ctx.AwaitEvent(
                eventName: $"agent-approval:{task.TenantId}:{task.IssueId}:{correlationId}",
                stepName: $"wait-for-human-review-{attempt}"
            );

            isApproved = review["approved"]?.GetValue<bool>() ?? false;
            lastFeedback = review["reason"]?.GetValue<string>() ?? "No feedback has been given";

            if (!isApproved)
            {
                _logger.LogWarning("Attempt {attempt} for tenant {TenantId} has been rejected: {reason}", attempt, task.TenantId, lastFeedback);
            }
        }

        if (isApproved)
        {
            _logger.LogInformation("Fix approved for tenant {TenantId}. Creating Pull Request...", task.TenantId);

            string prUrl = await ctx.Step("create-pull-request", async () =>
            {
                return await _gitHubService.CreatePullRequestAsync(task.TenantId, task.IssueId, "apply-fix", ctx.CancellationToken);
            });

            _logger.LogInformation("Mission accomplished for tenant {TenantId}, the PR has been created: {Url}", task.TenantId, prUrl);

            return new AgentResult { Success = true, PullRequestUrl = prUrl };
        }
        else
        {
            _logger.LogError("Maximum attempts reached. Escalates ticket {IssueId} for tenant {TenantId} to a human.", task.IssueId, task.TenantId);

            await ctx.Step("notify-senior-developer", async () =>
            {
                await _gitHubService.EscalateToSeniorAsync(task.TenantId, task.IssueId, "Agent didn't find a solution after 3 attempts.", ctx.CancellationToken);
            });

            return new AgentResult { Success = false, Reason = "Escalated to human supervisor after 3 failures." };
        }
    }
}

An ASP.NET Core Minimal API hosts and wires up SqlFlowSdk (with PostgreSQL persistence and NATS signaling) to run a background AutonomousAgentJob worker, while exposing two webhook endpoints to dispatch agent jobs and submit human-in-the-loop review approvals.

using Microsoft.AspNetCore.Mvc;
using SqlFlowSdk.Core;
using SqlFlowSdk.Postgres;
using SqlFlowSdk.Nats;
using SqlFlowSdk.DistributedSample.WorkerApp.Services;
using SqlFlowSdk.DistributedSample.WorkerApp.Models;

var builder = WebApplication.CreateBuilder(args);

builder.Configuration
    .AddJsonFile("appsettings.json", optional: false, reloadOnChange: true)
    .AddEnvironmentVariables();

// Read connection strings from configuration (appsettings.json or environment variables)
string pgConnectionString = builder.Configuration.GetConnectionString("SqlFlowDb")
    ?? throw new InvalidOperationException("Connection string 'SqlFlowDb' fehlt.");

// Read NATS connection string from configuration (appsettings.json or environment variables)
string natsUrl = builder.Configuration.GetConnectionString("nats-broker")
    ?? throw new InvalidOperationException("Connection string 'nats-broker' fehlt.");

builder.Services.AddLogging(loggingBuilder => loggingBuilder.AddConsole());

// Register Sample Services for LLM, GitHub, and Local Notifications
builder.Services.AddSingleton<ILlmService, LlmService>();
builder.Services.AddSingleton<IGitHubService, GitHubService>();
builder.Services.AddSingleton<ILocalNotificationService, LocalNotificationService>();

// Multitenant Configuration: Each tenant has its own queue for the autonomous agent
const string tenantId = "tenant-alpha";
const string queueName = "ai-agent-queue";

// Add SqlFlow Postgres and NATS Signaling, and configure the Worker with the AutonomousAgentJob
builder.Services
    .AddSqlFlowPostgres(pgConnectionString)
    .AddNatsSignaling(natsUrl)
    .AddWorker(tenantId, queueName, worker =>
    {
        worker
            .SetConcurrency(1)
            .SetPollInterval(1);

        worker.AddJob<AutonomousAgentJob, AgentTask, AgentResult>("solve-bug", options =>
        {
            options.WithMaxAttempts(3);
        });
    });

var app = builder.Build();

// API Endpoint to start the autonomous agent for a specific tenant and issue
app.MapPost("/agent/{tenantId}/start", async (
    [FromServices] IJobPublisher jobPublisher,
    [FromRoute] string tenantId,
    [FromBody] AgentTask task,
    CancellationToken ct) =>
{
    task.TenantId = tenantId;
    var result = await jobPublisher.PublishAsync<AutonomousAgentJob, AgentTask>(tenantId, "solve-bug", task, ct);

    return Results.Ok(new
    {
        TenantId = tenantId,
        RunId = result.RunId,
        TaskId = result.TaskId,
        Status = $"Agent dispatched for tenant {tenantId} to fix Issue #{task.IssueId}"
    });
});

// API Endpoint to receive human approval or rejection for the proposed fix
app.MapPost("/agent/{tenantId}/review/{issueId}/{correlationId}", async (
    [FromServices] IEventPublisher publisher,
    [FromRoute] string tenantId,
    [FromRoute] string issueId,
    [FromRoute] string correlationId,
    [FromBody] HumanApproval approval,
    CancellationToken ct) =>
{
    await publisher.EmitEventAsync(tenantId, queueName, eventName: $"agent-approval:{tenantId}:{issueId}:{correlationId}", payload: approval, ct);

    string message = approval.Approved
        ? $"Fix for tenant {tenantId}, correlation {correlationId} approved. Agent is now completing its work."
        : $"Fix for tenant {tenantId}, correlation {correlationId} rejected. Agent tries again with feedback: '{approval.Reason}'";

    return Results.Ok(new { Message = message });
});

app.Run();

Does it work?

There's a set of HTTP requests as a script available at:

We can execute these requests with ijhttp and see the results:

PS C:\Users\philipp\source\repos\bytefish\SqlFlowCore\requests> ijhttp --insecure --private-env-file .\http-client.env.json --env dotnet -L VERBOSE agent-requests.http
┌─────────────────────────────────────────────────────────────────────────────┐
│                      Running IntelliJ HTTP Client with                      │
├────────────────────────┬────────────────────────────────────────────────────┤
│         Files          │ agent-requests.http                                │
├────────────────────────┼────────────────────────────────────────────────────┤
│   Public Environment   │                                                    │
├────────────────────────┼────────────────────────────────────────────────────┤
│  Private Environment   │ baseUrl                                            │
└────────────────────────┴────────────────────────────────────────────────────┘
Request 'startAgent' POST http://localhost:5000/agent/tenant-alpha/start
= request =>
POST {{**baseUrl**}}/agent/tenant-alpha/start
Content-Type: application/json
Content-Length: 25
User-Agent: IntelliJ HTTP Client/CLI 2026.1
Accept-Encoding: br, deflate, gzip, x-gzip
Accept: */*

{
  "issue_id": "12345"
}

###
<= response =
HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8
Date: Sat, 14 Nov 2026 13:05:20 GMT
Server: Kestrel
Transfer-Encoding: chunked
{"tenantId":"tenant-alpha","runId":"726b079c-c596-4975-97b0-22337d431f81","taskId":"0f7008f3-06b5-4c96-9a20-f63b3628b6c0","status":"Agent dispatched for tenant tenant-alpha to fix Issue #12345"}
Response code: 200 (OK); Time: 545ms (545 ms); Content length: 194 bytes (194 B)
Request 'Reject the first attempt after a delay' POST http://localhost:5000/agent/tenant-alpha/review/12345/0f7008f3-06b5-4c96-9a20-f63b3628b6c0-attempt-1
= request =>
POST {{**baseUrl**}}/agent/tenant-alpha/review/12345/0f7008f3-06b5-4c96-9a20-f63b3628b6c0-attempt-1
Content-Type: application/json
Content-Length: 100
User-Agent: IntelliJ HTTP Client/CLI 2026.1
Accept-Encoding: br, deflate, gzip, x-gzip
Accept: */*

{
  "approved": false,
  "reason": "This is way too simple, add a better error handling strategy!"
}

###
<= response =
HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8
Date: Sat, 14 Nov 2026 13:05:52 GMT
Server: Kestrel
Transfer-Encoding: chunked
{"message":"Fix for tenant tenant-alpha, correlation 0f7008f3-06b5-4c96-9a20-f63b3628b6c0-attempt-1 rejected. Agent tries again with feedback: 'This is way too simple, add a better error handling strategy!'"}
Response code: 200 (OK); Time: 54ms (54 ms); Content length: 208 bytes (208 B)
Request 'Approve the second attempt after another delay' POST http://localhost:5000/agent/tenant-alpha/review/12345/0f7008f3-06b5-4c96-9a20-f63b3628b6c0-attempt-2
= request =>
POST {{**baseUrl**}}/agent/tenant-alpha/review/12345/0f7008f3-06b5-4c96-9a20-f63b3628b6c0-attempt-2
Content-Type: application/json
Content-Length: 59
User-Agent: IntelliJ HTTP Client/CLI 2026.1
Accept-Encoding: br, deflate, gzip, x-gzip
Accept: */*

{
  "approved": true,
  "reason": "Now, this looks good!"
}

###
<= response =
HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8
Date: Sat, 14 Nov 2026 13:06:22 GMT
Server: Kestrel
Transfer-Encoding: chunked
{"message":"Fix for tenant tenant-alpha, correlation 0f7008f3-06b5-4c96-9a20-f63b3628b6c0-attempt-2 approved. Agent is now completing its work."}
Response code: 200 (OK); Time: 77ms (77 ms); Content length: 145 bytes (145 B)



3 requests completed, 0 have failed tests
RUN SUCCESSFUL

Checking the CitusDB Cluster

Now, let's see which shard our data has ended upon.

Connect to the Coordinator node and run:

SELECT 
    r.tenant_id,
    r.run_id,
    r.state,
    get_shard_id_for_distribution_column('ssf.runs', r.tenant_id) AS shard_id,
    n.nodename AS worker_node,
    n.nodeport
FROM ssf.runs r
JOIN pg_dist_placement p 
    ON p.shardid = get_shard_id_for_distribution_column('ssf.runs', r.tenant_id)
JOIN pg_dist_node n 
    ON p.groupid = n.groupid
ORDER BY r.created_at DESC;

And we can see the data is put on worker-2:

tenant_id run_id state shard_id worker_node nodeport
tenant-alpha 726b079c-c596-4975-97b0-22337d431f81 completed 102052 worker-2 5432

Conclusion

I think it's been easy to horizontally scale the system using CitusDB.

And that's it for now!

How to contribute

One of the easiest ways to contribute is to participate in discussions. You can also contribute by submitting pull requests.

General feedback and discussions?

Do you have questions or feedback on this article? Please create an issue on the Repositories issue tracker.

Something is wrong or missing?

There may be something wrong or missing in this article. If you want to help fixing it, then please make a Pull Request to this file.