[AZ-1124] Add PT-10 gRPC stream perf scenario

Co-authored-by: Cursor <cursoragent@cursor.com>

SatelliteProvider.IntegrationTests/GrpcTestHelpers.cs

@@ -1,3 +1,5 @@
+using System.Net.Security;
+using System.Security.Cryptography.X509Certificates;
 using Grpc.Core;
 using Grpc.Net.Client;
 using Satellite.V1;
@@ -8,15 +10,65 @@ public static class GrpcTestHelpers
 {
     public static GrpcChannel CreateChannel(string apiUrl)
     {
+        var handler = new SocketsHttpHandler
+        {
+            EnableMultipleHttp2Connections = true,
+        };
+
+        var caCertPath = ResolveCaCertPath();
+        if (!string.IsNullOrEmpty(caCertPath))
+        {
+            var caCert = X509Certificate2.CreateFromPemFile(caCertPath);
+            handler.SslOptions = new SslClientAuthenticationOptions
+            {
+                RemoteCertificateValidationCallback = (_, certificate, _, errors) =>
+                {
+                    if (errors == SslPolicyErrors.None)
+                    {
+                        return true;
+                    }
+
+                    if (certificate is null)
+                    {
+                        return false;
+                    }
+
+                    using var chain = new X509Chain();
+                    chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust;
+                    chain.ChainPolicy.CustomTrustStore.Add(caCert);
+                    chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
+                    return chain.Build(new X509Certificate2(certificate));
+                },
+            };
+        }
+
         return GrpcChannel.ForAddress(apiUrl, new GrpcChannelOptions
         {
-            HttpHandler = new SocketsHttpHandler
-            {
-                EnableMultipleHttp2Connections = true,
-            },
+            HttpHandler = handler,
         });
     }
 
+    private static string? ResolveCaCertPath()
+    {
+        var explicitPath = Environment.GetEnvironmentVariable("PERF_CA_CERT");
+        if (!string.IsNullOrWhiteSpace(explicitPath) && File.Exists(explicitPath))
+        {
+            return explicitPath;
+        }
+
+        var projectRoot = Environment.GetEnvironmentVariable("PROJECT_ROOT");
+        if (!string.IsNullOrWhiteSpace(projectRoot))
+        {
+            var defaultPath = Path.Combine(projectRoot, "certs", "api.crt");
+            if (File.Exists(defaultPath))
+            {
+                return defaultPath;
+            }
+        }
+
+        return null;
+    }
+
     public static RouteTileDelivery.RouteTileDeliveryClient CreateClient(string apiUrl, string jwtToken)
     {
         var channel = CreateChannel(apiUrl);

SatelliteProvider.IntegrationTests/PerfBootstrap.cs

@@ -1,4 +1,7 @@
+using System.Diagnostics;
 using System.Security.Claims;
+using Grpc.Core;
+using Satellite.V1;
 using SatelliteProvider.TestSupport;
 using SixLabors.ImageSharp;
 using SixLabors.ImageSharp.Formats.Jpeg;
@@ -94,4 +97,207 @@ internal static class PerfBootstrap
         image.Save(stream, new JpegEncoder { Quality = 95 });
         return stream.ToArray();
     }
+
+    private static bool TryResolvePerfJwt(out string token)
+    {
+        var preMinted = Environment.GetEnvironmentVariable("PERF_JWT_TOKEN");
+        if (!string.IsNullOrWhiteSpace(preMinted))
+        {
+            token = preMinted;
+            return true;
+        }
+
+        try
+        {
+            var secret = JwtTestHelpers.ResolveSecretOrThrow();
+            var issuer = JwtTestHelpers.ResolveIssuerOrThrow();
+            var audience = JwtTestHelpers.ResolveAudienceOrThrow();
+            token = JwtTokenFactory.Create(
+                secret,
+                PerfSubject,
+                PerfTokenLifetime,
+                new[] { new Claim(PermissionsClaimType, GpsPermission) },
+                issuer: issuer,
+                audience: audience);
+            return true;
+        }
+        catch (InvalidOperationException ex)
+        {
+            Console.Error.WriteLine($"--run-pt10: {ex.Message}");
+            token = string.Empty;
+            return false;
+        }
+    }
+
+    public static async Task<int> RunPt10Async()
+    {
+        var apiUrl = Environment.GetEnvironmentVariable("API_URL") ?? "https://localhost:18980";
+        var repeatCount = int.TryParse(Environment.GetEnvironmentVariable("PERF_REPEAT_COUNT"), out var n) ? n : 20;
+        var slowMs = int.TryParse(Environment.GetEnvironmentVariable("PERF_PT10_SLOW_MS"), out var slow) ? slow : 50;
+
+        if (repeatCount < 1)
+        {
+            Console.Error.WriteLine("--run-pt10: PERF_REPEAT_COUNT must be >= 1");
+            return 1;
+        }
+
+        string token;
+        if (!TryResolvePerfJwt(out token))
+        {
+            return 1;
+        }
+
+        RouteTileDelivery.RouteTileDeliveryClient client;
+        try
+        {
+            client = GrpcTestHelpers.CreateClient(apiUrl, token);
+        }
+        catch (Exception ex)
+        {
+            Console.Error.WriteLine($"--run-pt10: failed to open gRPC channel to {apiUrl}: {ex.Message}");
+            return 1;
+        }
+
+        var headers = GrpcTestHelpers.AuthMetadata(token);
+        var firstBatchTimes = new List<long>();
+        var totalStreamTimes = new List<long>();
+        var failed = 0;
+
+        for (var i = 0; i < repeatCount; i++)
+        {
+            var result = await ProbeStreamAsync(client, headers, Guid.NewGuid(), delayMsBetweenEvents: null);
+            if (result.Success)
+            {
+                firstBatchTimes.Add(result.FirstBatchMs);
+                totalStreamTimes.Add(result.TotalStreamMs);
+                Console.Error.WriteLine($"  iteration {i + 1}/{repeatCount}: first_batch_ms={result.FirstBatchMs} total_stream_ms={result.TotalStreamMs}");
+            }
+            else
+            {
+                failed++;
+                Console.Error.WriteLine($"  iteration {i + 1}/{repeatCount}: FAILED — {result.FailureReason}");
+            }
+        }
+
+        if (firstBatchTimes.Count == 0)
+        {
+            Console.Error.WriteLine("--run-pt10: zero successful iterations");
+            Console.Out.WriteLine("PT10_ITERATIONS_OK=0");
+            Console.Out.WriteLine($"PT10_ITERATIONS_FAILED={failed}");
+            Console.Out.WriteLine("PT10_SLOW_CONSUMER=FAIL");
+            return 1;
+        }
+
+        var slowResult = await ProbeStreamAsync(client, headers, Guid.NewGuid(), delayMsBetweenEvents: slowMs);
+        var slowPass = slowResult.Success && slowResult.TileCount >= 1;
+
+        Console.Out.WriteLine($"PT10_FIRST_BATCH_P50={Percentile(firstBatchTimes, 50)}");
+        Console.Out.WriteLine($"PT10_FIRST_BATCH_P95={Percentile(firstBatchTimes, 95)}");
+        Console.Out.WriteLine($"PT10_TOTAL_STREAM_P50={Percentile(totalStreamTimes, 50)}");
+        Console.Out.WriteLine($"PT10_TOTAL_STREAM_P95={Percentile(totalStreamTimes, 95)}");
+        Console.Out.WriteLine($"PT10_ITERATIONS_OK={firstBatchTimes.Count}");
+        Console.Out.WriteLine($"PT10_ITERATIONS_FAILED={failed}");
+        Console.Out.WriteLine($"PT10_SLOW_CONSUMER={(slowPass ? "PASS" : "FAIL")}");
+        if (!slowPass)
+        {
+            Console.Error.WriteLine($"--run-pt10: slow-consumer check failed — {slowResult.FailureReason ?? "no tiles received"}");
+            return 1;
+        }
+
+        return failed > 0 ? 1 : 0;
+    }
+
+    internal static long Percentile(IReadOnlyList<long> values, int pct)
+    {
+        if (values.Count == 0)
+        {
+            return 0;
+        }
+
+        var sorted = values.OrderBy(v => v).ToList();
+        var idx = (int)Math.Ceiling(sorted.Count * pct / 100.0) - 1;
+        idx = Math.Clamp(idx, 0, sorted.Count - 1);
+        return sorted[idx];
+    }
+
+    private static async Task<StreamProbeResult> ProbeStreamAsync(
+        RouteTileDelivery.RouteTileDeliveryClient client,
+        Metadata headers,
+        Guid routeId,
+        int? delayMsBetweenEvents)
+    {
+        var request = GrpcTestHelpers.BuildValidRequest(routeId: routeId);
+        var sw = Stopwatch.StartNew();
+        long firstBatchMs = -1;
+        var tileCount = 0;
+        DeliveryComplete? complete = null;
+        DeliveryError? error = null;
+
+        try
+        {
+            using var call = client.DeliverRouteTiles(request, headers);
+            await foreach (var evt in call.ResponseStream.ReadAllAsync())
+            {
+                switch (evt.PayloadCase)
+                {
+                    case RouteTileEvent.PayloadOneofCase.Batch:
+                        if (firstBatchMs < 0)
+                        {
+                            firstBatchMs = sw.ElapsedMilliseconds;
+                        }
+
+                        tileCount += evt.Batch.Tiles.Count;
+                        break;
+                    case RouteTileEvent.PayloadOneofCase.Complete:
+                        complete = evt.Complete;
+                        break;
+                    case RouteTileEvent.PayloadOneofCase.Error:
+                        error = evt.Error;
+                        break;
+                }
+
+                if (delayMsBetweenEvents.HasValue)
+                {
+                    await Task.Delay(delayMsBetweenEvents.Value);
+                }
+            }
+        }
+        catch (RpcException ex)
+        {
+            return StreamProbeResult.Fail($"RpcException {ex.StatusCode}: {ex.Status.Detail}");
+        }
+
+        var totalStreamMs = sw.ElapsedMilliseconds;
+
+        if (error is not null)
+        {
+            return StreamProbeResult.Fail($"DeliveryError {error.Code}: {error.Message}");
+        }
+
+        if (complete is null)
+        {
+            return StreamProbeResult.Fail("stream ended without DeliveryComplete");
+        }
+
+        if (firstBatchMs < 0)
+        {
+            return StreamProbeResult.Fail("no Batch event before DeliveryComplete");
+        }
+
+        return StreamProbeResult.Ok(firstBatchMs, totalStreamMs, tileCount);
+    }
+
+    private readonly record struct StreamProbeResult(
+        bool Success,
+        long FirstBatchMs,
+        long TotalStreamMs,
+        int TileCount,
+        string? FailureReason)
+    {
+        public static StreamProbeResult Ok(long firstBatchMs, long totalStreamMs, int tileCount) =>
+            new(true, firstBatchMs, totalStreamMs, tileCount, null);
+
+        public static StreamProbeResult Fail(string reason) =>
+            new(false, 0, 0, 0, reason);
+    }
 }

SatelliteProvider.IntegrationTests/Program.cs

@@ -19,6 +19,10 @@ class Program
             {
                 return PerfBootstrap.GenerateUavFixture(args);
             }
+            if (args[0].Equals("--run-pt10", StringComparison.OrdinalIgnoreCase))
+            {
+                return await PerfBootstrap.RunPt10Async();
+            }
         }
 
         var apiUrl = Environment.GetEnvironmentVariable("API_URL") ?? "https://api:8080";

SatelliteProvider.IntegrationTests/SatelliteProvider.IntegrationTests.csproj

@@ -19,4 +19,8 @@
     <ProjectReference Include="..\SatelliteProvider.Common\SatelliteProvider.Common.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <InternalsVisibleTo Include="SatelliteProvider.Tests" />
+  </ItemGroup>
+
 </Project>