Performance Optimizations

Overview

Performance Optimizations transforms the Release Orchestrator into a high-performance system capable of handling enterprise-scale deployments. This enhancement provides parallel gate evaluation, bulk digest resolution, agent task batching, optimized database queries, and intelligent caching strategies.

This is a best-in-class implementation focused on reducing latency, increasing throughput, and ensuring the system scales efficiently under load.

Design Principles

Measure First: Optimize based on profiling data, not assumptions
Parallel by Default: Concurrent execution where dependencies allow
Cache Intelligently: Cache at the right level with proper invalidation
Batch Operations: Reduce round-trips through batching
Async Everything: Non-blocking operations throughout
Graceful Degradation: Performance degrades linearly, not exponentially

Architecture

Component Overview

┌────────────────────────────────────────────────────────────────────────┐
│                   Performance Optimization System                      │
├────────────────────────────────────────────────────────────────────────┤
│                                                                        │
│  ┌──────────────────┐    ┌───────────────────┐    ┌─────────────────┐ │
│  │ ParallelGate     │    │ BulkDigestResolver│    │ QueryOptimizer  │ │
│  │ Evaluator        │    │                   │    │                 │ │
│  └──────────────────┘    └───────────────────┘    └─────────────────┘ │
│           │                       │                        │          │
│           ▼                       ▼                        ▼          │
│  ┌──────────────────┐    ┌───────────────────┐    ┌─────────────────┐ │
│  │ TaskBatcher      │    │ CacheManager      │    │ ConnectionPool  │ │
│  │                  │    │                   │    │                 │ │
│  └──────────────────┘    └───────────────────┘    └─────────────────┘ │
│           │                       │                        │          │
│           ▼                       ▼                        ▼          │
│  ┌──────────────────┐    ┌───────────────────┐    ┌─────────────────┐ │
│  │ Prefetcher       │    │ IndexManager      │    │ LoadBalancer    │ │
│  │                  │    │                   │    │                 │ │
│  └──────────────────┘    └───────────────────┘    └─────────────────┘ │
│                                                                        │
└────────────────────────────────────────────────────────────────────────┘

Key Components

1. ParallelGateEvaluator

Evaluates multiple gates concurrently:

public sealed class ParallelGateEvaluator
{
    private readonly ImmutableArray<IGateEvaluator> _evaluators;
    private readonly SemaphoreSlim _concurrencyLimiter;
    private readonly IGateResultCache _cache;

    public ParallelGateEvaluator(ParallelGateConfig config)
    {
        _concurrencyLimiter = new SemaphoreSlim(config.MaxConcurrentEvaluations);
    }

    public async Task<GateEvaluationResult> EvaluateAllAsync(
        PromotionContext context,
        IReadOnlyList<GateDefinition> gates,
        CancellationToken ct)
    {
        var result = new GateEvaluationResult
        {
            PromotionId = context.PromotionId,
            StartedAt = _timeProvider.GetUtcNow()
        };

        // Group gates by dependency
        var executionPlan = BuildExecutionPlan(gates);

        foreach (var stage in executionPlan.Stages)
        {
            // Execute all gates in this stage concurrently
            var stageTasks = stage.Gates.Select(async gate =>
            {
                await _concurrencyLimiter.WaitAsync(ct);
                try
                {
                    return await EvaluateSingleGateAsync(gate, context, ct);
                }
                finally
                {
                    _concurrencyLimiter.Release();
                }
            });

            var stageResults = await Task.WhenAll(stageTasks);
            result.GateResults.AddRange(stageResults);

            // Check for failures that should stop evaluation
            var failures = stageResults.Where(r => r.Status == GateStatus.Failed && r.Gate.StopOnFailure);
            if (failures.Any())
            {
                result.Status = GateEvaluationStatus.Failed;
                result.FailedGates = failures.Select(f => f.Gate.Id).ToImmutableArray();
                break;
            }
        }

        result.CompletedAt = _timeProvider.GetUtcNow();
        return result;
    }

    private async Task<SingleGateResult> EvaluateSingleGateAsync(
        GateDefinition gate,
        PromotionContext context,
        CancellationToken ct)
    {
        // Check cache first
        var cacheKey = BuildCacheKey(gate, context);
        var cached = await _cache.GetAsync(cacheKey, ct);
        if (cached != null && !IsExpired(cached, gate.CacheTtl))
        {
            return cached with { FromCache = true };
        }

        // Evaluate
        var evaluator = _evaluators.First(e => e.CanEvaluate(gate.Type));
        var sw = Stopwatch.StartNew();

        try
        {
            var result = await evaluator.EvaluateAsync(gate, context, ct);
            sw.Stop();

            result = result with
            {
                EvaluationDuration = sw.Elapsed,
                EvaluatedAt = _timeProvider.GetUtcNow()
            };

            // Cache result
            await _cache.SetAsync(cacheKey, result, gate.CacheTtl, ct);

            return result;
        }
        catch (Exception ex)
        {
            return new SingleGateResult
            {
                GateId = gate.Id,
                Status = GateStatus.Error,
                Error = ex.Message,
                EvaluationDuration = sw.Elapsed
            };
        }
    }

    private GateExecutionPlan BuildExecutionPlan(IReadOnlyList<GateDefinition> gates)
    {
        var plan = new GateExecutionPlan();
        var remaining = gates.ToList();
        var completed = new HashSet<Guid>();

        while (remaining.Any())
        {
            // Find gates with all dependencies satisfied
            var ready = remaining
                .Where(g => g.DependsOn.All(d => completed.Contains(d)))
                .ToList();

            if (!ready.Any())
            {
                throw new CircularDependencyException(remaining.Select(g => g.Id));
            }

            plan.Stages.Add(new GateExecutionStage { Gates = ready.ToImmutableArray() });

            foreach (var gate in ready)
            {
                completed.Add(gate.Id);
                remaining.Remove(gate);
            }
        }

        return plan;
    }
}

2. BulkDigestResolver

Resolves multiple image digests in parallel:

public sealed class BulkDigestResolver
{
    private readonly IRegistryClientPool _clientPool;
    private readonly IDigestCache _cache;
    private readonly int _maxConcurrency;

    public async Task<IReadOnlyDictionary<string, string>> ResolveAllAsync(
        IReadOnlyList<ImageReference> images,
        CancellationToken ct)
    {
        var results = new ConcurrentDictionary<string, string>();

        // Check cache first
        var uncached = new List<ImageReference>();
        foreach (var image in images)
        {
            var cached = await _cache.GetAsync(image.FullReference, ct);
            if (cached != null)
            {
                results[image.FullReference] = cached;
            }
            else
            {
                uncached.Add(image);
            }
        }

        if (!uncached.Any())
        {
            return results.ToImmutableDictionary();
        }

        // Group by registry for connection reuse
        var byRegistry = uncached.GroupBy(i => i.Registry);

        await Parallel.ForEachAsync(
            byRegistry,
            new ParallelOptions { MaxDegreeOfParallelism = _maxConcurrency, CancellationToken = ct },
            async (group, ct) =>
            {
                var client = await _clientPool.GetClientAsync(group.Key, ct);
                try
                {
                    // Batch resolve for this registry
                    var digests = await client.ResolveDigestsAsync(
                        group.Select(i => (i.Repository, i.Tag)).ToList(), ct);

                    foreach (var (image, digest) in group.Zip(digests))
                    {
                        results[image.FullReference] = digest;
                        await _cache.SetAsync(image.FullReference, digest, _cacheTtl, ct);
                    }
                }
                finally
                {
                    _clientPool.ReturnClient(client);
                }
            });

        return results.ToImmutableDictionary();
    }
}

public interface IRegistryClient
{
    // Single resolution
    Task<string> ResolveDigestAsync(string repository, string tag, CancellationToken ct);

    // Batch resolution (more efficient)
    Task<IReadOnlyList<string>> ResolveDigestsAsync(
        IReadOnlyList<(string Repository, string Tag)> images,
        CancellationToken ct);
}

3. TaskBatcher

Batches agent tasks for efficiency:

public sealed class TaskBatcher
{
    private readonly ConcurrentDictionary<Guid, TaskBatch> _batches = new();
    private readonly TimeSpan _batchWindow;
    private readonly int _maxBatchSize;

    public async Task<Guid> EnqueueAsync(
        AgentTask task,
        CancellationToken ct)
    {
        var agentId = task.TargetAgentId;

        // Get or create batch for this agent
        var batch = _batches.GetOrAdd(agentId, _ => new TaskBatch
        {
            AgentId = agentId,
            CreatedAt = _timeProvider.GetUtcNow(),
            Tasks = new ConcurrentBag<AgentTask>()
        });

        batch.Tasks.Add(task);

        // Check if batch should be sent
        if (ShouldFlushBatch(batch))
        {
            await FlushBatchAsync(agentId, ct);
        }

        return batch.Id;
    }

    private bool ShouldFlushBatch(TaskBatch batch)
    {
        // Flush if max size reached
        if (batch.Tasks.Count >= _maxBatchSize)
            return true;

        // Flush if batch window expired
        if (_timeProvider.GetUtcNow() - batch.CreatedAt >= _batchWindow)
            return true;

        // Flush if high-priority task added
        if (batch.Tasks.Any(t => t.Priority == TaskPriority.Immediate))
            return true;

        return false;
    }

    private async Task FlushBatchAsync(Guid agentId, CancellationToken ct)
    {
        if (!_batches.TryRemove(agentId, out var batch))
            return;

        var tasks = batch.Tasks.ToArray();
        if (!tasks.Any())
            return;

        _logger.LogDebug(
            "Flushing batch of {Count} tasks to agent {AgentId}",
            tasks.Length, agentId);

        // Group tasks by type for optimized execution
        var grouped = tasks.GroupBy(t => t.TaskType);

        foreach (var group in grouped)
        {
            var batchedPayload = CreateBatchedPayload(group.ToList());
            await _agentClient.SendBatchAsync(agentId, batchedPayload, ct);
        }
    }

    private BatchedTaskPayload CreateBatchedPayload(IReadOnlyList<AgentTask> tasks)
    {
        // Optimize payload based on task type
        return tasks.First().TaskType switch
        {
            TaskType.Deploy => CreateDeployBatch(tasks),
            TaskType.HealthCheck => CreateHealthCheckBatch(tasks),
            TaskType.WriteSticker => CreateStickerBatch(tasks),
            _ => CreateGenericBatch(tasks)
        };
    }

    private BatchedTaskPayload CreateDeployBatch(IReadOnlyList<AgentTask> tasks)
    {
        // Deduplicate image pulls
        var uniqueImages = tasks
            .SelectMany(t => t.Payload.Images)
            .Distinct()
            .ToList();

        return new BatchedTaskPayload
        {
            Type = BatchType.Deploy,
            Images = uniqueImages,      // Pull once, deploy many
            Tasks = tasks.Select(t => new SlimTaskPayload
            {
                TaskId = t.Id,
                ContainerName = t.Payload.ContainerName,
                ImageIndex = uniqueImages.IndexOf(t.Payload.Image)
            }).ToImmutableArray()
        };
    }
}

4. CacheManager

Multi-level caching with intelligent invalidation:

public sealed class CacheManager
{
    private readonly IMemoryCache _l1Cache;          // In-process
    private readonly IDistributedCache _l2Cache;     // Redis
    private readonly ICacheInvalidator _invalidator;

    public async Task<T?> GetOrSetAsync<T>(
        string key,
        Func<CancellationToken, Task<T>> factory,
        CacheOptions options,
        CancellationToken ct) where T : class
    {
        // L1 check
        if (_l1Cache.TryGetValue(key, out T? l1Value))
        {
            _metrics.RecordHit("l1");
            return l1Value;
        }

        // L2 check
        var l2Value = await _l2Cache.GetAsync<T>(key, ct);
        if (l2Value != null)
        {
            _metrics.RecordHit("l2");

            // Populate L1
            _l1Cache.Set(key, l2Value, new MemoryCacheEntryOptions
            {
                AbsoluteExpirationRelativeToNow = options.L1Ttl,
                Size = EstimateSize(l2Value)
            });

            return l2Value;
        }

        // Cache miss - compute value
        _metrics.RecordMiss();
        var value = await factory(ct);

        if (value != null)
        {
            // Set L1
            _l1Cache.Set(key, value, new MemoryCacheEntryOptions
            {
                AbsoluteExpirationRelativeToNow = options.L1Ttl,
                Size = EstimateSize(value)
            });

            // Set L2
            await _l2Cache.SetAsync(key, value, new DistributedCacheEntryOptions
            {
                AbsoluteExpirationRelativeToNow = options.L2Ttl
            }, ct);

            // Register for invalidation
            if (options.InvalidationTags != null)
            {
                await _invalidator.RegisterAsync(key, options.InvalidationTags, ct);
            }
        }

        return value;
    }

    public async Task InvalidateByTagAsync(string tag, CancellationToken ct)
    {
        var keys = await _invalidator.GetKeysByTagAsync(tag, ct);

        foreach (var key in keys)
        {
            _l1Cache.Remove(key);
            await _l2Cache.RemoveAsync(key, ct);
        }

        await _invalidator.UnregisterTagAsync(tag, ct);
    }
}

public sealed record CacheOptions
{
    public TimeSpan L1Ttl { get; init; } = TimeSpan.FromMinutes(5);
    public TimeSpan L2Ttl { get; init; } = TimeSpan.FromHours(1);
    public ImmutableArray<string>? InvalidationTags { get; init; }
    public bool AllowStale { get; init; }
}

5. QueryOptimizer

Optimizes database queries:

public sealed class QueryOptimizer
{
    public async Task<IReadOnlyList<Release>> GetReleasesOptimizedAsync(
        ReleaseQuery query,
        CancellationToken ct)
    {
        // Build optimized query
        var sql = new StringBuilder();
        sql.AppendLine(@"
            SELECT r.*,
                   c.name as component_name, c.digest as component_digest,
                   e.name as env_name, e.status as env_status
            FROM releases r");

        // Use indexed join strategy based on query
        if (query.EnvironmentId.HasValue)
        {
            // Use environment index
            sql.AppendLine(@"
                INNER JOIN release_environments re ON r.id = re.release_id
                    AND re.environment_id = @EnvironmentId");
        }

        sql.AppendLine(@"
            LEFT JOIN release_components c ON r.id = c.release_id
            LEFT JOIN environments e ON r.current_environment_id = e.id
            WHERE r.tenant_id = @TenantId");

        // Apply filters with index hints
        if (query.Status.HasValue)
        {
            sql.AppendLine("AND r.status = @Status");  // Uses idx_releases_status
        }

        if (query.CreatedAfter.HasValue)
        {
            sql.AppendLine("AND r.created_at >= @CreatedAfter");  // Uses idx_releases_created
        }

        // Optimized ordering
        sql.AppendLine("ORDER BY r.created_at DESC");

        // Pagination with keyset (faster than OFFSET)
        if (query.Cursor != null)
        {
            sql.AppendLine("AND r.created_at < @CursorCreatedAt");
            sql.AppendLine("AND r.id < @CursorId");
        }

        sql.AppendLine("LIMIT @Limit");

        // Execute with read replica if available
        var connection = query.AllowStale
            ? await _connectionPool.GetReadReplicaAsync(ct)
            : await _connectionPool.GetPrimaryAsync(ct);

        return await connection.QueryAsync<Release>(sql.ToString(), query, ct);
    }

    public void EnsureIndexes()
    {
        // Ensure critical indexes exist
        var requiredIndexes = new[]
        {
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_releases_tenant_status ON releases(tenant_id, status)",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_releases_tenant_created ON releases(tenant_id, created_at DESC)",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_releases_env ON releases(current_environment_id) WHERE current_environment_id IS NOT NULL",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_components_release ON release_components(release_id)",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_deployments_release ON deployments(release_id, created_at DESC)",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_promotions_release ON promotions(release_id, status)",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_evidence_subject ON evidence_packets(subject_id, subject_type)"
        };

        foreach (var index in requiredIndexes)
        {
            _migrationRunner.EnsureIndex(index);
        }
    }
}

6. Prefetcher

Proactively loads data:

public sealed class Prefetcher
{
    public async Task PrefetchForPromotionAsync(
        Guid releaseId,
        Guid targetEnvironmentId,
        CancellationToken ct)
    {
        // Prefetch in parallel
        var tasks = new List<Task>
        {
            // Release and components
            _releaseCache.WarmAsync(releaseId, ct),

            // Target environment
            _environmentCache.WarmAsync(targetEnvironmentId, ct),

            // Gates for this environment
            _gateCache.WarmForEnvironmentAsync(targetEnvironmentId, ct),

            // Recent scan results
            _scanCache.WarmForReleaseAsync(releaseId, ct),

            // Approval policies
            _policyCache.WarmForEnvironmentAsync(targetEnvironmentId, ct),

            // Available agents
            _agentCache.WarmForEnvironmentAsync(targetEnvironmentId, ct)
        };

        await Task.WhenAll(tasks);
    }

    public async Task PrefetchForDashboardAsync(
        Guid tenantId,
        CancellationToken ct)
    {
        // Predictive prefetch based on user behavior
        var recentQueries = await _queryHistoryStore.GetRecentAsync(tenantId, ct);
        var predictedQueries = _predictor.Predict(recentQueries);

        foreach (var query in predictedQueries.Take(10))
        {
            _ = ExecuteAndCacheAsync(query, ct);  // Fire and forget
        }
    }
}

7. ConnectionPool

Optimized connection management:

public sealed class ConnectionPool
{
    private readonly ObjectPool<NpgsqlConnection> _primaryPool;
    private readonly ObjectPool<NpgsqlConnection> _replicaPool;
    private readonly ILoadBalancer _replicaBalancer;

    public async Task<PooledConnection> GetPrimaryAsync(CancellationToken ct)
    {
        var connection = _primaryPool.Get();
        if (connection.State != ConnectionState.Open)
        {
            await connection.OpenAsync(ct);
        }
        return new PooledConnection(connection, () => _primaryPool.Return(connection));
    }

    public async Task<PooledConnection> GetReadReplicaAsync(CancellationToken ct)
    {
        // Select replica based on load
        var replica = _replicaBalancer.SelectReplica();

        var connection = _replicaPool.Get();
        connection.ConnectionString = replica.ConnectionString;

        if (connection.State != ConnectionState.Open)
        {
            await connection.OpenAsync(ct);
        }

        return new PooledConnection(connection, () => _replicaPool.Return(connection));
    }

    public void WarmPool()
    {
        // Pre-create connections
        Parallel.For(0, _config.MinPoolSize, _ =>
        {
            var connection = new NpgsqlConnection(_config.ConnectionString);
            connection.Open();
            _primaryPool.Return(connection);
        });
    }
}

public sealed class PooledConnection : IAsyncDisposable
{
    private readonly NpgsqlConnection _connection;
    private readonly Action _returnAction;

    public PooledConnection(NpgsqlConnection connection, Action returnAction)
    {
        _connection = connection;
        _returnAction = returnAction;
    }

    public NpgsqlConnection Connection => _connection;

    public async ValueTask DisposeAsync()
    {
        _returnAction();
    }
}

Performance Benchmarks

Target Metrics

Operation	Current	Target	Optimization
Gate evaluation (5 gates)	5s (sequential)	1.5s (parallel)	ParallelGateEvaluator
Digest resolution (10 images)	10s	2s	BulkDigestResolver
Promotion creation	500ms	100ms	Prefetching
Dashboard load	2s	500ms	Caching + Query optimization
Deployment start	3s	500ms	Task batching
Agent task throughput	100/s	1000/s	Connection pooling

Load Test Scenarios

public sealed class PerformanceTests
{
    [Fact]
    public async Task Gate_Evaluation_Should_Complete_Under_Target()
    {
        // Arrange
        var gates = CreateGates(count: 10);
        var context = CreatePromotionContext();

        // Act
        var sw = Stopwatch.StartNew();
        var result = await _evaluator.EvaluateAllAsync(context, gates, CancellationToken.None);
        sw.Stop();

        // Assert
        Assert.True(sw.Elapsed < TimeSpan.FromSeconds(2));
        Assert.Equal(GateEvaluationStatus.Succeeded, result.Status);
    }

    [Fact]
    public async Task Concurrent_Promotions_Should_Scale_Linearly()
    {
        // Test with 1, 10, 50, 100 concurrent promotions
        var results = new List<(int Count, TimeSpan Duration)>();

        foreach (var count in new[] { 1, 10, 50, 100 })
        {
            var promotions = Enumerable.Range(0, count)
                .Select(_ => CreatePromotionRequest())
                .ToList();

            var sw = Stopwatch.StartNew();
            await Task.WhenAll(promotions.Select(p =>
                _promotionService.CreateAsync(p, CancellationToken.None)));
            sw.Stop();

            results.Add((count, sw.Elapsed));
        }

        // Assert linear scaling (within 2x factor)
        var baseline = results[0].Duration.TotalMilliseconds;
        foreach (var (count, duration) in results.Skip(1))
        {
            var expectedMax = baseline * count * 2;
            Assert.True(duration.TotalMilliseconds < expectedMax,
                $"Count {count}: {duration.TotalMilliseconds}ms exceeded {expectedMax}ms");
        }
    }
}

Configuration

Performance Tuning Options

performance:
  # Gate evaluation
  gates:
    max_concurrent_evaluations: 10
    evaluation_timeout: "00:00:30"
    cache_ttl: "00:05:00"

  # Digest resolution
  digest_resolution:
    max_concurrent_registries: 5
    max_concurrent_per_registry: 10
    cache_ttl: "01:00:00"
    timeout: "00:00:30"

  # Task batching
  task_batching:
    enabled: true
    batch_window: "00:00:01"
    max_batch_size: 50

  # Caching
  cache:
    l1:
      enabled: true
      max_size_mb: 256
      default_ttl: "00:05:00"
    l2:
      enabled: true
      provider: redis  # Valkey (Redis-compatible)
      connection_string: "redis://localhost:6379"
      default_ttl: "01:00:00"

  # Database
  database:
    primary:
      min_pool_size: 10
      max_pool_size: 100
      connection_timeout: "00:00:05"
    read_replicas:
      enabled: true
      hosts:
        - host: replica1.db.local
          weight: 50
        - host: replica2.db.local
          weight: 50
      load_balancing: round_robin

  # Prefetching
  prefetch:
    enabled: true
    promotion_warmup: true
    dashboard_prediction: true
    prediction_depth: 10

  # Connection pooling
  http_client:
    max_connections_per_host: 100
    connection_lifetime: "00:05:00"
    keep_alive_timeout: "00:00:30"

  # gRPC
  grpc:
    max_concurrent_streams: 100
    keepalive_time: "00:01:00"
    keepalive_timeout: "00:00:20"

Metrics & Observability

Prometheus Metrics

# Latency histograms
stella_gate_evaluation_duration_seconds{gate_type}
stella_digest_resolution_duration_seconds{registry}
stella_promotion_creation_duration_seconds
stella_deployment_start_duration_seconds

# Cache metrics
stella_cache_hits_total{level, cache}
stella_cache_misses_total{cache}
stella_cache_size_bytes{level, cache}
stella_cache_evictions_total{cache, reason}

# Connection pools
stella_connection_pool_size{pool}
stella_connection_pool_active{pool}
stella_connection_pool_wait_seconds{pool}

# Batching
stella_batch_size{operation}
stella_batch_flush_total{operation, reason}
stella_batch_latency_seconds{operation}

# Query performance
stella_query_duration_seconds{query_type}
stella_query_rows_returned{query_type}
stella_index_scan_total{table, index}

# Throughput
stella_operations_per_second{operation}
stella_concurrent_operations{operation}

API Design

Performance-Optimized Endpoints

# Batch operations
POST   /api/v1/batch/digests              # Bulk digest resolution
POST   /api/v1/batch/releases             # Bulk release creation
POST   /api/v1/batch/gates                # Parallel gate evaluation

# Prefetch hints
POST   /api/v1/prefetch/promotion         # Warm cache for promotion
POST   /api/v1/prefetch/dashboard         # Warm cache for dashboard

# Cache management
DELETE /api/v1/cache/invalidate           # Invalidate cache entries
GET    /api/v1/cache/stats                # Cache statistics

# Health & metrics
GET    /api/v1/performance/stats          # Performance statistics
GET    /api/v1/performance/slow-queries   # Recent slow queries