Tweaks to load tool

Author: rbev

tools/testing-tool/TestingTool.AppHost/AppHost.cs

@@ -3,7 +3,7 @@
 using TestingTool.AppHost;
 
 var options = CliOptions.Parse(args);
-var persistenceType = options.GetEnumOrDefault<PersistenceType>("persistence", PersistenceType.PostgreSql);
+var persistenceType = options.GetEnumOrDefault("persistence", PersistenceType.PostgreSql);
 Console.WriteLine($"Using persistence type: {persistenceType}");
 
 var builder = DistributedApplication.CreateBuilder(args);
@@ -26,6 +26,7 @@
 
 var errorInstance = platform
     .AddServiceControlErrorInstance("error", raven)
+    .WithEnvironment("SERVICECONTROL_DISABLEEXTERNALINTEGRATIONSPUBLISHING", "true")
     .WithEnvironment("OTEL_EXPORTER_OTLP_ENDPOINT", observability.Collector.GetEndpoint("otlp-grpc"))
     .WithPersistenceType(persistenceType)
     .WithRunMode(PlatformRunMode.SetupAndRun);

tools/testing-tool/TestingTool.Contracts/BypassStatus.cs

@@ -18,6 +18,11 @@ public sealed class BypassStatus
     /// <summary>Total failed-message envelopes written directly to the error queue since process start.</summary>
     public long ErrorsWritten { get; init; }
 
+    /// <summary>Total bypass sends that failed since process start. A non-zero value indicates
+    /// transport/broker issues; see logs (Warning level) for details. Previously these failures
+    /// were swallowed at Debug level, making the bypass appear idle when sends were failing.</summary>
+    public long ErrorsFailed { get; init; }
+
     /// <summary>When the current bypass run started (UTC ISO 8601, null if idle).</summary>
     public string? StartedAt { get; init; }
 }

tools/testing-tool/TestingTool.Contracts/StartBypassRequest.cs

@@ -15,4 +15,12 @@ public sealed class StartBypassRequest
 
     /// <summary>Optional auto-stop duration in seconds. Null/0 = run until explicitly stopped.</summary>
     public double? DurationSeconds { get; init; }
+
+    /// <summary>
+    /// Number of parallel worker tasks to use for sending. Each task runs its own timer at
+    /// <c>rate / parallelism</c> msg/s so the aggregate approaches the target rate. Defaults to
+    /// <see cref="Environment.ProcessorCount"/> when omitted. Increase this if the bypass is
+    /// not hitting the target rate due to send latency on a single thread.
+    /// </summary>
+    public int? Parallelism { get; init; }
 }

tools/testing-tool/TestingTool.Contracts/TestingToolStatus.cs

@@ -24,6 +24,10 @@ public sealed class TestingToolStatus
     /// <summary>Total errors emitted via the direct error-queue bypass writer since process start.</summary>
     public long BypassErrorsWritten { get; init; }
 
+    /// <summary>Total bypass sends that failed since process start. Non-zero values indicate
+    /// transport/broker issues on the bypass path — the handler path may be unaffected.</summary>
+    public long BypassErrorsFailed { get; init; }
+
     /// <summary>The shard id this replica owns, used for disjoint scenario slices when scaled out.</summary>
     public string? ShardId { get; init; }
 

tools/testing-tool/TestingTool/DirectErrorQueueWriter.cs

@@ -33,14 +33,16 @@ public sealed class DirectErrorQueueWriter
     private readonly Counter<long> _bypassCounter;
 
     private CancellationTokenSource? _cts;
-    private Task? _loop;
+    private Task[]? _loops;
     private long _errorsWritten;
+    private long _errorsFailed;
     private double _currentRate;
     private string? _activeScenario;
     private DateTimeOffset _startedAt;
 
     public bool IsRunning => _cts is not null;
     public long ErrorsWritten => Interlocked.Read(ref _errorsWritten);
+    public long ErrorsFailed => Interlocked.Read(ref _errorsFailed);
     public double CurrentRate => _currentRate;
     public string? ActiveScenario => _activeScenario;
 
@@ -62,7 +64,7 @@ public DirectErrorQueueWriter(
     }
 
     /// <summary>Starts writing failed-message envelopes directly to the error queue.</summary>
-    public bool TryStart(string scenarioName, double rate, TimeSpan? duration, out string? error)
+    public bool TryStart(string scenarioName, double rate, TimeSpan? duration, int? parallelism, out string? error)
     {
         var scenario = _registry.Get(scenarioName);
         if (scenario is null)
@@ -83,6 +85,11 @@ public bool TryStart(string scenarioName, double rate, TimeSpan? duration, out s
             return false;
         }
 
+        // Default to ProcessorCount workers. Each worker runs its own timer at rate/N so the
+        // aggregate approaches the target. More workers = more concurrent sends = higher
+        // throughput when individual sends have latency.
+        var workerCount = parallelism is { } p and > 0 ? p : Environment.ProcessorCount;
+
         _activeScenario = scenarioName;
         _currentRate = rate;
         _startedAt = DateTimeOffset.UtcNow;
@@ -92,10 +99,15 @@ public bool TryStart(string scenarioName, double rate, TimeSpan? duration, out s
             : new CancellationTokenSource();
         _cts = cts;
 
-        _loop = Task.Run(() => WriteLoop(scenario, rate, cts.Token));
+        _loops = new Task[workerCount];
+        for (var i = 0; i < workerCount; i++)
+        {
+            var workerIndex = i;
+            _loops[i] = Task.Run(() => WriteLoop(scenario, rate / workerCount, workerIndex, cts.Token));
+        }
 
-        _logger.LogInformation("Started bypass writer for scenario {Scenario} at {Rate:F1} msg/s{Duration}",
-            scenarioName, rate, duration is null ? "" : $" for {duration.Value}");
+        _logger.LogInformation("Started bypass writer for scenario {Scenario} at {Rate:F1} msg/s{Duration} across {Workers} workers",
+            scenarioName, rate, duration is null ? "" : $" for {duration.Value}", workerCount);
 
         error = null;
         return true;
@@ -107,10 +119,27 @@ public void Stop()
         if (_cts is null) return;
 
         _cts.Cancel();
+
+        // Await all worker loops before disposing the token so we don't dispose a CTS that's still
+        // in flight inside _session.Send. Swallow the expected cancellation/timeout.
+        var loops = _loops;
+        if (loops is not null)
+        {
+            try
+            {
+                Task.WaitAll(loops, TimeSpan.FromSeconds(5));
+            }
+            catch
+            {
+                // Loops were cancelled or timed out — expected during stop.
+            }
+        }
+
         _cts.Dispose();
         _cts = null;
+        _loops = null;
 
-        _logger.LogInformation("Stopped bypass writer after {Errors} errors written", ErrorsWritten);
+        _logger.LogInformation("Stopped bypass writer: {Errors} written, {Failed} failed", ErrorsWritten, ErrorsFailed);
 
         _activeScenario = null;
         _currentRate = 0;
@@ -123,14 +152,16 @@ public void Stop()
         Scenario = _activeScenario,
         Rate = _currentRate,
         ErrorsWritten = ErrorsWritten,
+        ErrorsFailed = ErrorsFailed,
         StartedAt = IsRunning ? _startedAt.ToString("O") : null
     };
 
     /// <summary>
     /// The load generation loop: sends <see cref="LoadMessage"/> directly to the error queue
-    /// with failure headers at the target rate until cancelled.
+    /// with failure headers at the target rate until cancelled. Multiple instances run in
+    /// parallel, each handling a fraction of the total rate.
     /// </summary>
-    private async Task WriteLoop(IScenario scenario, double rate, CancellationToken ct)
+    private async Task WriteLoop(IScenario scenario, double rate, int workerIndex, CancellationToken ct)
     {
         var interval = TimeSpan.FromSeconds(1.0 / rate);
         using var timer = new PeriodicTimer(interval);
@@ -184,9 +215,20 @@ private async Task WriteLoop(IScenario scenario, double rate, CancellationToken
                     _metrics.AddBypassErrorsWritten(1);
                     _bypassCounter.Add(1, new KeyValuePair<string, object?>("scenario", scenario.Name));
                 }
-                catch (Exception ex) when (ex is not OperationCanceledException)
+                catch (OperationCanceledException)
                 {
-                    _logger.LogDebug(ex, "Bypass send failed for scenario {Scenario} seq {Seq}", scenario.Name, seq);
+                    // Cancellation is expected on stop/timeout — let it propagate to the outer handler.
+                    throw;
+                }
+                catch (Exception ex)
+                {
+                    Interlocked.Increment(ref _errorsFailed);
+                    _metrics.AddBypassErrorsFailed(1);
+
+                    // Log at Warning so send failures are visible in default logging configs.
+                    // Previously this was LogDebug, which silently swallowed transport/broker
+                    // failures and made the bypass appear idle when sends were actually failing.
+                    _logger.LogWarning(ex, "Bypass send failed for scenario {Scenario} worker {Worker} seq {Seq}", scenario.Name, workerIndex, seq);
                 }
             }
         }

tools/testing-tool/TestingTool/Program.cs

@@ -103,6 +103,7 @@
     ErrorsArchived = metrics.TotalErrorsArchived,
     SearchesExecuted = metrics.TotalSearches,
     BypassErrorsWritten = metrics.TotalBypassErrorsWritten,
+    BypassErrorsFailed = metrics.TotalBypassErrorsFailed,
     ShardId = shardId,
     ActiveScenarios = metrics.ActiveScenarios,
     ActiveJobs = jobRunner.GetSnapshot().Count(j => j.Running),
@@ -192,7 +193,7 @@
         ? TimeSpan.FromSeconds(secs)
         : (TimeSpan?)null;
 
-    if (!writer.TryStart(scenarioName, rate, duration, out var error))
+    if (!writer.TryStart(scenarioName, rate, duration, request?.Parallelism, out var error))
         return Results.BadRequest(new { error });
 
     return Results.Ok(writer.GetStatus());

tools/testing-tool/TestingTool/TestingToolMetrics.cs

@@ -12,6 +12,7 @@ public sealed class TestingToolMetrics
     private long _totalErrorsArchived;
     private long _totalSearches;
     private long _totalBypassErrorsWritten;
+    private long _totalBypassErrorsFailed;
     private long _activeScenarios;
     private double _currentRate;
 
@@ -20,6 +21,7 @@ public sealed class TestingToolMetrics
     public long TotalErrorsArchived => Interlocked.Read(ref _totalErrorsArchived);
     public long TotalSearches => Interlocked.Read(ref _totalSearches);
     public long TotalBypassErrorsWritten => Interlocked.Read(ref _totalBypassErrorsWritten);
+    public long TotalBypassErrorsFailed => Interlocked.Read(ref _totalBypassErrorsFailed);
     public int ActiveScenarios => (int)Interlocked.Read(ref _activeScenarios);
     public double CurrentRate => _currentRate;
 
@@ -28,6 +30,7 @@ public sealed class TestingToolMetrics
     public void AddErrorsArchived(long count) => Interlocked.Add(ref _totalErrorsArchived, count);
     public void AddSearches(long count) => Interlocked.Add(ref _totalSearches, count);
     public void AddBypassErrorsWritten(long count) => Interlocked.Add(ref _totalBypassErrorsWritten, count);
+    public void AddBypassErrorsFailed(long count) => Interlocked.Add(ref _totalBypassErrorsFailed, count);
     public void SetActiveScenarios(int count) => Interlocked.Exchange(ref _activeScenarios, count);
     public void SetCurrentRate(double rate) => _currentRate = rate;
 }

tools/testing-tool/TestingTool/wwwroot/index.html

@@ -234,8 +234,12 @@ <h2>Direct Error-Queue Bypass</h2>
         </select>
         <span class="field-label">Rate (msg/s):</span>
         <input type="number" id="bypass-rate" value="100" min="1" step="1" />
+        <span class="field-label">Workers:</span>
+        <input type="number" id="bypass-parallelism" value="" min="1" step="1" placeholder="auto" title="Number of parallel worker tasks (default: CPU count)" />
         <span class="field-label">Duration (sec, optional):</span>
         <input type="number" id="bypass-duration" value="" min="1" step="1" placeholder="∞" />
+      </div>
+      <div class="controls">
         <button class="btn start" onclick="startBypass()" id="btn-bypass-start">▶ Start Bypass</button>
         <button class="btn stop" onclick="stopBypass()" id="btn-bypass-stop" disabled>⏹ Stop Bypass</button>
       </div>
@@ -554,8 +558,10 @@ <h2>Direct Error-Queue Bypass</h2>
       const scenario = document.getElementById('bypass-scenario').value;
       const rate = parseFloat(document.getElementById('bypass-rate').value);
       const dur = parseInt(document.getElementById('bypass-duration').value);
+      const par = parseInt(document.getElementById('bypass-parallelism').value);
       const body = { scenario };
       if (rate && rate > 0) body.rate = rate;
+      if (par && par > 0) body.parallelism = par;
       if (dur && dur > 0) body.durationSeconds = dur;
       try {
         await api('POST', '/api/bypass/start', body);
@@ -592,6 +598,7 @@ <h2>Direct Error-Queue Bypass</h2>
           ${bypass.scenario ? `<span>Scenario: <strong>${bypass.scenario}</strong></span>` : ''}
           ${bypass.running ? `<span>Rate: <strong>${bypass.rate ?? 0} msg/s</strong></span>` : ''}
           <span>Errors written: <strong>${fmtNum(bypass.errorsWritten)}</strong></span>
+          ${bypass.errorsFailed ? `<span style="color:var(--err)">Failed: <strong>${fmtNum(bypass.errorsFailed)}</strong></span>` : ''}
         `;
       } catch (e) { /* ignore */ }
     }