From 5469408944c4800b3c87922823d5202389f59558 Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Mon, 19 Oct 2026 11:17:36 +0800 Subject: [PATCH 1/2] Pin pacakge in test project to avoid CVE --- .../testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj | 1 + 1 file changed, 1 insertion(+) diff --git a/tools/testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj b/tools/testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj index 36bc67fa14..16769d8545 100644 --- a/tools/testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj +++ b/tools/testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj @@ -20,6 +20,7 @@ + From 40fc20e1eccf4143f56c072aad3cf73314902be7 Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Mon, 19 Oct 2026 13:39:49 +0800 Subject: [PATCH 2/2] Set defaults for ingestion concurrency --- docs/ingestion-pipeline.md | 25 +++++++------------ .../When_hosting_error_ingestion_only.cs | 4 +-- .../Auditing/AuditIngestion.cs | 2 +- .../Infrastructure/Settings/Settings.cs | 4 +-- .../Ingestion/IngestionSettingsReaderTests.cs | 25 +++++++++++++++++-- .../Ingestion/IngestionSettingsReader.cs | 22 +++++++++++++--- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../PostgreSqlTransportCustomization.cs | 4 --- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../SqlServerTransportCustomization.cs | 4 --- .../ServiceControlAuditEndpointTests.cs | 2 +- .../ServiceControlPrimaryEndpointTests.cs | 2 +- .../TransportCustomization.cs | 4 +-- .../TransportSettings.cs | 5 ++++ .../HostApplicationBuilderExtensions.cs | 2 +- .../Infrastructure/Settings/Settings.cs | 4 +-- .../Operations/ErrorIngestion.cs | 2 +- 37 files changed, 90 insertions(+), 65 deletions(-) diff --git a/docs/ingestion-pipeline.md b/docs/ingestion-pipeline.md index d747db318d..57b70c1d78 100644 --- a/docs/ingestion-pipeline.md +++ b/docs/ingestion-pipeline.md @@ -46,9 +46,9 @@ Each instance reads its settings from its own settings root | Setting | Default | Range | What it does | | --- | --- | --- | --- | -| `ErrorIngestionBatchSize` / `AuditIngestionBatchSize` | the transport's `MaximumConcurrencyLevel` | 1 to 1000 | The most messages one write handles | -| `ErrorIngestionMaxParallelWriters` / `AuditIngestionMaxParallelWriters` | the storage decides, see below | 1 to 16 | How many batches are written at once | -| `ErrorIngestionBatchTimeout` / `AuditIngestionBatchTimeout` | `00:00:00` | 0 to 5 seconds | How long a batch that is not yet full waits for more messages | +| `ErrorIngestionBatchSize` / `AuditIngestionBatchSize` | `25`, or the transport's `MaximumConcurrencyLevel` if that is lower | 1 to 1000 | The most messages one write handles | +| `ErrorIngestionMaxParallelWriters` / `AuditIngestionMaxParallelWriters` | `4` where the storage takes concurrent batches, otherwise `1`, see below | 1 to 16 | How many batches are written at once | +| `ErrorIngestionBatchTimeout` / `AuditIngestionBatchTimeout` | `00:00:00.100` | 0 to 5 seconds | How long a batch that is not yet full waits for more messages | As environment variables: @@ -64,21 +64,13 @@ SERVICECONTROL_AUDIT_AuditIngestionBatchTimeout=00:00:00.100 ### Batch size -The default, the transport's concurrency, is also the ceiling. A receive does not return until the -batch carrying its message has been written, so the transport holds every one of its concurrency -slots open and no more than that many messages can ever be waiting. Setting a batch size above the -transport's concurrency is therefore unreachable, with or without a batch timeout: the batch never -fills, and a timeout only delays what has already arrived. Larger batches come from raising -`MaximumConcurrencyLevel`, and this setting is what holds writes below it when a storage is happier -with smaller ones. +The default batch size is 25, capped at the transport's concurrency so that a lower concurrency still allows full batches. A receive does not return until its batch has been written, so no more than that many messages can wait. A configured batch size above the transport's concurrency never fills, and a timeout only delays messages already received. + +Error and audit instances default to a concurrency of 100 for every transport unless `MaximumConcurrencyLevel` is configured. In the error instance, this shared setting also controls the hosted endpoint, retry forwarding, and throughput collection receivers. The monitoring instance has its own concurrency defaults. To allow batches larger than the receive concurrency, raise `MaximumConcurrencyLevel` and configure the batch size. ### Batch timeout -Zero means a partial batch is written without waiting, which is what the ingestion did before -the setting existed. A non-zero value trades latency for fewer, larger writes: at volume it costs -nothing, because a full batch never waits, and at a trickle it delays each message by up to the -timeout. Start at 100ms if a storage is clearly happier with larger batches. It cannot make a batch -larger than the transport's concurrency, only fuller. +The default is 100 ms. An explicit zero writes partial batches without waiting. A nonzero timeout trades latency for fewer, larger writes: full batches never wait, but partial batches wait up to the timeout. The default can therefore add up to 100 ms of latency at low message volumes. A timeout cannot make a batch larger than the transport's concurrency, only fuller. ### Parallel writers @@ -100,9 +92,10 @@ someone actually asked for rather than a default. | Error, SQL Server and PostgreSQL | yes | The batch writer was built for it: upserts guarded by the attempt times so the newer attempt wins whichever transaction commits last, inserts that tolerate a competing writer's identical row, and a consistent lock order. Running several `--error-ingestion-only` hosts against one database already depends on all of it. | | Error, RavenDB | no | Failed messages are merged by patch scripts that read and rewrite one document, and nothing orders two patches of the same document against each other. `--error-ingestion-only` refuses to start on RavenDB for the same reason. | | Audit, RavenDB | yes | Audit documents are independent. Nothing merges two of them, and every batch gets its own bulk insert operation. | +| Audit, SQL Server and PostgreSQL | yes | The audit unit of work factory supports concurrent batches. | | Audit, in memory | no | Test storage only. | -A storage that says yes gets four writers by default. +Error ingestion defaults to four writers on SQL Server and PostgreSQL and one on RavenDB. Audit ingestion defaults to four writers on all three providers. In-memory test storage uses one writer. ### The ordering that concurrency does not excuse diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/When_hosting_error_ingestion_only.cs b/src/ServiceControl.AcceptanceTests/Recoverability/When_hosting_error_ingestion_only.cs index 092652e7d9..6ec152e712 100644 --- a/src/ServiceControl.AcceptanceTests/Recoverability/When_hosting_error_ingestion_only.cs +++ b/src/ServiceControl.AcceptanceTests/Recoverability/When_hosting_error_ingestion_only.cs @@ -71,7 +71,7 @@ public async Task Should_ingest_without_an_endpoint_and_without_the_single_owner } [Test] - public async Task Should_default_the_concurrency_to_10_when_none_is_configured() + public async Task Should_default_the_concurrency_to_100_when_none_is_configured() { var settings = await CreateSettings(); settings.MaximumConcurrencyLevel = null; @@ -81,7 +81,7 @@ public async Task Should_default_the_concurrency_to_10_when_none_is_configured() try { Assert.DoesNotThrow(() => host.Services.GetServices().ToArray()); - Assert.That(host.Services.GetRequiredService().MaxConcurrency, Is.EqualTo(10)); + Assert.That(host.Services.GetRequiredService().MaxConcurrency, Is.EqualTo(100)); } finally { diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs index 533e46f2b4..4502d4e98f 100644 --- a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs +++ b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs @@ -45,7 +45,7 @@ ILogger logger throw new ArgumentException("MaxConcurrency is not set in TransportSettings"); } - MaxBatchSize = settings.AuditIngestionBatchSize ?? transportSettings.MaxConcurrency.Value; + MaxBatchSize = IngestionSettingsReader.ResolveBatchSize(settings.AuditIngestionBatchSize, transportSettings.MaxConcurrency.Value); pipeline = new IngestionPipeline( new IngestionPipelineSettings diff --git a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs index 7a6b8d978f..0c744efc81 100644 --- a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs +++ b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs @@ -189,7 +189,7 @@ public int MaxBodySizeToStore public TimeSpan TimeToRestartAuditIngestionAfterFailure { get; set; } /// - /// The most messages one write handles. Null leaves it to the transport's concurrency. + /// The most messages one write handles. Null is 25, or the transport's concurrency if that is lower. /// public int? AuditIngestionBatchSize { get; set; } @@ -200,7 +200,7 @@ public int MaxBodySizeToStore public int? AuditIngestionMaxParallelWriters { get; set; } /// - /// How long a batch that is not yet full waits for more messages. + /// How long a batch that is not yet full waits for more messages. Unset is 100 ms; zero does not wait. /// public TimeSpan AuditIngestionBatchTimeout { get; set; } diff --git a/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs index 065fb7c240..2d65cc1682 100644 --- a/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs +++ b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs @@ -19,13 +19,13 @@ class IngestionSettingsReaderTests public void ClearSetting() => Environment.SetEnvironmentVariable(EnvironmentVariable, null); [Test] - public void An_unconfigured_setting_leaves_the_choice_to_the_caller() + public void An_unconfigured_setting_is_left_to_the_caller_or_the_default() { using (Assert.EnterMultipleScope()) { Assert.That(IngestionSettingsReader.ReadBatchSize(Root, SettingName, validateConfiguration: true), Is.Null); Assert.That(IngestionSettingsReader.ReadMaxParallelWriters(Root, SettingName, validateConfiguration: true), Is.Null); - Assert.That(IngestionSettingsReader.ReadBatchTimeout(Root, SettingName, validateConfiguration: true), Is.EqualTo(TimeSpan.Zero)); + Assert.That(IngestionSettingsReader.ReadBatchTimeout(Root, SettingName, validateConfiguration: true), Is.EqualTo(TimeSpan.FromMilliseconds(100))); } } @@ -37,6 +37,27 @@ public void A_configured_setting_is_read() Assert.That(IngestionSettingsReader.ReadBatchSize(Root, SettingName, validateConfiguration: true), Is.EqualTo(250)); } + [Test] + public void A_configured_zero_timeout_is_kept() + { + Configure("00:00:00"); + + Assert.That(IngestionSettingsReader.ReadBatchTimeout(Root, SettingName, validateConfiguration: true), Is.EqualTo(TimeSpan.Zero)); + } + + [TestCase(1, 1)] + [TestCase(10, 10)] + [TestCase(25, 25)] + [TestCase(26, 25)] + [TestCase(100, 25)] + public void An_unconfigured_batch_size_is_25_capped_at_the_transport_concurrency(int concurrency, int expected) => + Assert.That(IngestionSettingsReader.ResolveBatchSize(null, concurrency), Is.EqualTo(expected)); + + [TestCase(10)] + [TestCase(100)] + public void A_configured_batch_size_is_kept(int concurrency) => + Assert.That(IngestionSettingsReader.ResolveBatchSize(50, concurrency), Is.EqualTo(50)); + [Test] public void A_configured_timeout_is_read() { diff --git a/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs b/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs index 8ffcdd24a3..3d2fa8e848 100644 --- a/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs +++ b/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs @@ -13,8 +13,7 @@ namespace ServiceControl.Infrastructure.Ingestion; public static class IngestionSettingsReader { /// - /// The most messages one write handles. Absent leaves it to the transport's concurrency, which - /// is as many messages as can be waiting to be written at any one time. + /// The most messages one write handles. Absent leaves it to . /// public static int? ReadBatchSize(SettingsRootNamespace settingsNamespace, string name, bool validateConfiguration) => ReadInt(settingsNamespace, name, validateConfiguration, minimum: 1, maximum: MaximumBatchSize); @@ -28,13 +27,14 @@ public static class IngestionSettingsReader ReadInt(settingsNamespace, name, validateConfiguration, minimum: 1, maximum: MaximumParallelWriters); /// - /// How long a batch that is not yet full waits for more messages. Absent does not wait at all. + /// How long a batch that is not yet full waits for more messages. Absent waits + /// ; an explicit zero does not wait at all. /// public static TimeSpan ReadBatchTimeout(SettingsRootNamespace settingsNamespace, string name, bool validateConfiguration) { if (!SettingsReader.TryRead(settingsNamespace, name, out var value)) { - return TimeSpan.Zero; + return DefaultBatchTimeout; } if (!TimeSpan.TryParse(value, out var timeout)) @@ -50,6 +50,13 @@ public static TimeSpan ReadBatchTimeout(SettingsRootNamespace settingsNamespace, return timeout; } + /// + /// Settles how many messages one write handles. A batch larger than the transport's concurrency + /// never fills, so the default is capped at it and a lower concurrency stays attainable. + /// + public static int ResolveBatchSize(int? configured, int transportMaxConcurrency) => + configured ?? Math.Min(DefaultBatchSize, transportMaxConcurrency); + /// /// Settles how many writers a pipeline actually gets. A storage whose batches are not safe to /// interleave holds it at one whatever is configured, and says so when that overrules a @@ -100,6 +107,13 @@ static Exception Invalid(string message) /// public const int DefaultMaxParallelWriters = 4; + /// + /// The most messages one write handles when nothing is configured and the transport allows it. + /// + public const int DefaultBatchSize = 25; + + public static readonly TimeSpan DefaultBatchTimeout = TimeSpan.FromMilliseconds(100); + const int MaximumBatchSize = 1000; const int MaximumParallelWriters = 16; static readonly TimeSpan MaximumBatchTimeout = TimeSpan.FromSeconds(5); diff --git a/src/ServiceControl.Transports.ASBS.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.ASBS.Tests/ServiceControlAuditEndpointTests.cs index 20c547e296..78da3c372c 100644 --- a/src/ServiceControl.Transports.ASBS.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.ASBS.Tests/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.ASBS.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.ASBS.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.ASBS.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.ASBS.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.ASQ.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.ASQ.Tests/ServiceControlAuditEndpointTests.cs index 20c547e296..78da3c372c 100644 --- a/src/ServiceControl.Transports.ASQ.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.ASQ.Tests/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.ASQ.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.ASQ.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.ASQ.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.ASQ.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.IBMMQ.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.IBMMQ.Tests/ServiceControlAuditEndpointTests.cs index f70a724658..453e78e736 100644 --- a/src/ServiceControl.Transports.IBMMQ.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.IBMMQ.Tests/ServiceControlAuditEndpointTests.cs @@ -4,5 +4,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.IBMMQ.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.IBMMQ.Tests/ServiceControlPrimaryEndpointTests.cs index 56579e59cd..a0a73cc855 100644 --- a/src/ServiceControl.Transports.IBMMQ.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.IBMMQ.Tests/ServiceControlPrimaryEndpointTests.cs @@ -4,5 +4,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.Msmq.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.Msmq.Tests/ServiceControlAuditEndpointTests.cs index 20c547e296..78da3c372c 100644 --- a/src/ServiceControl.Transports.Msmq.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.Msmq.Tests/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.Msmq.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.Msmq.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.Msmq.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.Msmq.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.PostgreSql.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.PostgreSql.Tests/ServiceControlAuditEndpointTests.cs index 197afd5970..78da3c372c 100644 --- a/src/ServiceControl.Transports.PostgreSql.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.PostgreSql.Tests/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.PostgreSql.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.PostgreSql.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.PostgreSql.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.PostgreSql.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs b/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs index 063c32b3e4..7abb2b9b47 100644 --- a/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs +++ b/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs @@ -15,15 +15,11 @@ public class PostgreSqlTransportCustomization() : TransportCustomization 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.RabbitMQClassicConventionalRouting.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.RabbitMQClassicConventionalRouting.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.RabbitMQClassicConventionalRouting.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.RabbitMQClassicConventionalRouting.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.RabbitMQClassicDirectRouting.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.RabbitMQClassicDirectRouting.Tests/ServiceControlAuditEndpointTests.cs index 20c547e296..78da3c372c 100644 --- a/src/ServiceControl.Transports.RabbitMQClassicDirectRouting.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.RabbitMQClassicDirectRouting.Tests/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.RabbitMQClassicDirectRouting.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.RabbitMQClassicDirectRouting.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.RabbitMQClassicDirectRouting.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.RabbitMQClassicDirectRouting.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.RabbitMQQuorumConventionalRouting.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.RabbitMQQuorumConventionalRouting.Tests/ServiceControlAuditEndpointTests.cs index 20c547e296..78da3c372c 100644 --- a/src/ServiceControl.Transports.RabbitMQQuorumConventionalRouting.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.RabbitMQQuorumConventionalRouting.Tests/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.RabbitMQQuorumConventionalRouting.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.RabbitMQQuorumConventionalRouting.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.RabbitMQQuorumConventionalRouting.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.RabbitMQQuorumConventionalRouting.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.RabbitMQQuorumDirectRouting.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.RabbitMQQuorumDirectRouting.Tests/ServiceControlAuditEndpointTests.cs index 20c547e296..78da3c372c 100644 --- a/src/ServiceControl.Transports.RabbitMQQuorumDirectRouting.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.RabbitMQQuorumDirectRouting.Tests/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.RabbitMQQuorumDirectRouting.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.RabbitMQQuorumDirectRouting.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.RabbitMQQuorumDirectRouting.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.RabbitMQQuorumDirectRouting.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.SQS.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.SQS.Tests/ServiceControlAuditEndpointTests.cs index 20c547e296..78da3c372c 100644 --- a/src/ServiceControl.Transports.SQS.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.SQS.Tests/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.SQS.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.SQS.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.SQS.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.SQS.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.SqlServer.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.SqlServer.Tests/ServiceControlAuditEndpointTests.cs index 197afd5970..78da3c372c 100644 --- a/src/ServiceControl.Transports.SqlServer.Tests/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.SqlServer.Tests/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.SqlServer.Tests/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.SqlServer.Tests/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.SqlServer.Tests/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.SqlServer.Tests/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs b/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs index 35be6b3a11..c873ca6379 100644 --- a/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs +++ b/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs @@ -20,16 +20,12 @@ protected override void CustomizeTransportForPrimaryEndpoint(EndpointConfigurati #pragma warning disable CS0618 // Type or member is obsolete routing.EnableMessageDrivenPubSubCompatibilityMode(); #pragma warning restore CS0618 // Type or member is obsolete - - transportSettings.MaxConcurrency ??= 10; } //Do not EnableMessageDrivenPubSubCompatibilityMode for send-only endpoint protected override void CustomizeTransportForAuditEndpoint(EndpointConfiguration endpointConfiguration, SqlServerTransport transportDefinition, TransportSettings transportSettings) { transportDefinition.TransportTransactionMode = TransportTransactionMode.ReceiveOnly; - - transportSettings.MaxConcurrency ??= 10; } protected override void CustomizeTransportForMonitoringEndpoint(EndpointConfiguration endpointConfiguration, SqlServerTransport transportDefinition, TransportSettings transportSettings) diff --git a/src/ServiceControl.Transports.Tests/Learning/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.Tests/Learning/ServiceControlAuditEndpointTests.cs index 20c547e296..78da3c372c 100644 --- a/src/ServiceControl.Transports.Tests/Learning/ServiceControlAuditEndpointTests.cs +++ b/src/ServiceControl.Transports.Tests/Learning/ServiceControlAuditEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlAuditEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 32; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.Tests/Learning/ServiceControlPrimaryEndpointTests.cs b/src/ServiceControl.Transports.Tests/Learning/ServiceControlPrimaryEndpointTests.cs index 465bba2734..61355b073a 100644 --- a/src/ServiceControl.Transports.Tests/Learning/ServiceControlPrimaryEndpointTests.cs +++ b/src/ServiceControl.Transports.Tests/Learning/ServiceControlPrimaryEndpointTests.cs @@ -2,5 +2,5 @@ namespace ServiceControl.Transport.Tests; partial class ServiceControlPrimaryEndpointTests { - private static partial int GetTransportDefaultConcurrency() => 10; + private static partial int GetTransportDefaultConcurrency() => 100; } \ No newline at end of file diff --git a/src/ServiceControl.Transports/TransportCustomization.cs b/src/ServiceControl.Transports/TransportCustomization.cs index f7d4142eba..b2086f989f 100644 --- a/src/ServiceControl.Transports/TransportCustomization.cs +++ b/src/ServiceControl.Transports/TransportCustomization.cs @@ -82,7 +82,7 @@ public void CustomizePrimaryEndpoint(EndpointConfiguration endpointConfiguration var transport = CreateTransport(transportSettings); CustomizeTransportForPrimaryEndpoint(endpointConfiguration, transport, transportSettings); - transportSettings.MaxConcurrency ??= 10; + transportSettings.MaxConcurrency ??= TransportSettings.DefaultIngestionMaxConcurrency; endpointConfiguration.UseTransport(transport); } @@ -93,7 +93,7 @@ public void CustomizeAuditEndpoint(EndpointConfiguration endpointConfiguration, var transport = CreateTransport(transportSettings); CustomizeTransportForAuditEndpoint(endpointConfiguration, transport, transportSettings); - transportSettings.MaxConcurrency ??= 32; + transportSettings.MaxConcurrency ??= TransportSettings.DefaultIngestionMaxConcurrency; endpointConfiguration.SendOnly(); diff --git a/src/ServiceControl.Transports/TransportSettings.cs b/src/ServiceControl.Transports/TransportSettings.cs index 246879e482..a58089a8bf 100644 --- a/src/ServiceControl.Transports/TransportSettings.cs +++ b/src/ServiceControl.Transports/TransportSettings.cs @@ -15,6 +15,11 @@ public class TransportSettings : SettingsHolder public string EndpointName { get; set; } + /// + /// What the error and audit instances receive with when nothing is configured. + /// + public const int DefaultIngestionMaxConcurrency = 100; + public int? MaxConcurrency { get; set; } public bool RunCustomChecks { get; set; } diff --git a/src/ServiceControl/HostApplicationBuilderExtensions.cs b/src/ServiceControl/HostApplicationBuilderExtensions.cs index 6b3ac7c6e0..42c19edb51 100644 --- a/src/ServiceControl/HostApplicationBuilderExtensions.cs +++ b/src/ServiceControl/HostApplicationBuilderExtensions.cs @@ -115,7 +115,7 @@ public static void AddServiceControl(this IHostApplicationBuilder hostBuilder, S // Ingestion receives through its own transport infrastructure and forwards through // that same infrastructure's dispatcher, so the endpoint is not hosted at all. // A hosted endpoint gets this default from CustomizePrimaryEndpoint, which this branch skips. - transportSettings.MaxConcurrency ??= 10; + transportSettings.MaxConcurrency ??= TransportSettings.DefaultIngestionMaxConcurrency; var machineName = NServiceBus.Support.RuntimeEnvironment.MachineName; services.AddSingleton(new HostInformation( diff --git a/src/ServiceControl/Infrastructure/Settings/Settings.cs b/src/ServiceControl/Infrastructure/Settings/Settings.cs index fad7491699..3fb9ccd03f 100644 --- a/src/ServiceControl/Infrastructure/Settings/Settings.cs +++ b/src/ServiceControl/Infrastructure/Settings/Settings.cs @@ -210,7 +210,7 @@ public string InstanceId public TimeSpan TimeToRestartErrorIngestionAfterFailure { get; set; } /// - /// The most messages one write handles. Null leaves it to the transport's concurrency. + /// The most messages one write handles. Null is 25, or the transport's concurrency if that is lower. /// public int? ErrorIngestionBatchSize { get; set; } @@ -221,7 +221,7 @@ public string InstanceId public int? ErrorIngestionMaxParallelWriters { get; set; } /// - /// How long a batch that is not yet full waits for more messages. + /// How long a batch that is not yet full waits for more messages. Unset is 100 ms; zero does not wait. /// public TimeSpan ErrorIngestionBatchTimeout { get; set; } public int? MaximumConcurrencyLevel { get; set; } diff --git a/src/ServiceControl/Operations/ErrorIngestion.cs b/src/ServiceControl/Operations/ErrorIngestion.cs index d23623c6d4..76ad0bbc03 100644 --- a/src/ServiceControl/Operations/ErrorIngestion.cs +++ b/src/ServiceControl/Operations/ErrorIngestion.cs @@ -45,7 +45,7 @@ public ErrorIngestion( throw new ArgumentException("MaxConcurrency is not set in TransportSettings"); } - MaxBatchSize = settings.ErrorIngestionBatchSize ?? transportSettings.MaxConcurrency.Value; + MaxBatchSize = IngestionSettingsReader.ResolveBatchSize(settings.ErrorIngestionBatchSize, transportSettings.MaxConcurrency.Value); pipeline = new IngestionPipeline( new IngestionPipelineSettings