From 743d6ce09ce19a9aa65791ebc637550e975d068b Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Mon, 19 Oct 2026 13:39:49 +0800 Subject: [PATCH 1/4] 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 647b549024..bada2bb8b4 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 From 877d71fb2a7e87f1ecb95cf9eb1027224455aacc Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Wed, 21 Oct 2026 12:31:08 +0800 Subject: [PATCH 2/4] Revert change to transport max parallelism --- docs/ingestion-pipeline.md | 6 ++--- .../When_hosting_error_ingestion_only.cs | 4 +-- .../Auditing/AuditIngestion.cs | 5 ++-- .../Infrastructure/Settings/Settings.cs | 2 +- .../Ingestion/IngestionSettingsReaderTests.cs | 25 ++++++++++--------- .../Ingestion/IngestionSettingsReader.cs | 14 ++++------- .../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 | 2 +- .../Operations/ErrorIngestion.cs | 5 ++-- 37 files changed, 66 insertions(+), 64 deletions(-) diff --git a/docs/ingestion-pipeline.md b/docs/ingestion-pipeline.md index 57b70c1d78..71ba862e82 100644 --- a/docs/ingestion-pipeline.md +++ b/docs/ingestion-pipeline.md @@ -46,7 +46,7 @@ Each instance reads its settings from its own settings root | Setting | Default | Range | What it does | | --- | --- | --- | --- | -| `ErrorIngestionBatchSize` / `AuditIngestionBatchSize` | `25`, or the transport's `MaximumConcurrencyLevel` if that is lower | 1 to 1000 | The most messages one write handles | +| `ErrorIngestionBatchSize` / `AuditIngestionBatchSize` | the transport's `MaximumConcurrencyLevel` divided by the writers, at least 1 | 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 | @@ -64,9 +64,9 @@ SERVICECONTROL_AUDIT_AuditIngestionBatchTimeout=00:00:00.100 ### Batch size -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. +The default batch size is the transport's concurrency divided by the number of writers, using integer division and never less than 1. The writer count is the one the pipeline actually gets, so a storage that holds ingestion to one writer gets a batch as large as the concurrency. When the concurrency is below the writer count, the batch size is 1. A receive does not return until its batch has been written, so no more than the concurrency can wait. A configured batch size is used as given. One 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. +Without `MaximumConcurrencyLevel`, the error instance receives with a concurrency of 10. The audit instance defaults to 32, except SQL Server and PostgreSQL transports, which default to 10. In the error instance, this shared setting also controls the hosted endpoint, retry forwarding, and throughput collection receivers. To allow larger batches, raise `MaximumConcurrencyLevel`. ### Batch timeout 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 6ec152e712..092652e7d9 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_100_when_none_is_configured() + public async Task Should_default_the_concurrency_to_10_when_none_is_configured() { var settings = await CreateSettings(); settings.MaximumConcurrencyLevel = null; @@ -81,7 +81,7 @@ public async Task Should_default_the_concurrency_to_100_when_none_is_configured( try { Assert.DoesNotThrow(() => host.Services.GetServices().ToArray()); - Assert.That(host.Services.GetRequiredService().MaxConcurrency, Is.EqualTo(100)); + Assert.That(host.Services.GetRequiredService().MaxConcurrency, Is.EqualTo(10)); } finally { diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs index 4502d4e98f..fece955a25 100644 --- a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs +++ b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs @@ -45,13 +45,14 @@ ILogger logger throw new ArgumentException("MaxConcurrency is not set in TransportSettings"); } - MaxBatchSize = IngestionSettingsReader.ResolveBatchSize(settings.AuditIngestionBatchSize, transportSettings.MaxConcurrency.Value); + var writers = IngestionSettingsReader.ResolveMaxParallelWriters(settings.AuditIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.AuditIngestionMaxParallelWriters), logger); + MaxBatchSize = IngestionSettingsReader.ResolveBatchSize(settings.AuditIngestionBatchSize, transportSettings.MaxConcurrency.Value, writers); pipeline = new IngestionPipeline( new IngestionPipelineSettings { BatchSize = MaxBatchSize, - MaxWriters = IngestionSettingsReader.ResolveMaxParallelWriters(settings.AuditIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.AuditIngestionMaxParallelWriters), logger), + MaxWriters = writers, BatchTimeout = settings.AuditIngestionBatchTimeout }, IngestBatch, diff --git a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs index 0c744efc81..da096a3720 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 is 25, or the transport's concurrency if that is lower. + /// The most messages one write handles. Null splits the transport's concurrency across the writers. /// public int? AuditIngestionBatchSize { get; set; } diff --git a/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs index 2d65cc1682..e566097c44 100644 --- a/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs +++ b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs @@ -45,18 +45,19 @@ public void A_configured_zero_timeout_is_kept() 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)); + [TestCase(10, 1, 10)] + [TestCase(10, 4, 2)] + [TestCase(32, 4, 8)] + [TestCase(10, 3, 3)] + [TestCase(3, 4, 1)] + [TestCase(1, 4, 1)] + public void An_unconfigured_batch_size_divides_the_transport_concurrency_across_the_writers(int concurrency, int writers, int expected) => + Assert.That(IngestionSettingsReader.ResolveBatchSize(null, concurrency, writers), Is.EqualTo(expected)); + + [TestCase(10, 1)] + [TestCase(100, 4)] + public void A_configured_batch_size_is_kept(int concurrency, int writers) => + Assert.That(IngestionSettingsReader.ResolveBatchSize(50, concurrency, writers), 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 3d2fa8e848..0ced632aab 100644 --- a/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs +++ b/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs @@ -51,11 +51,12 @@ public static TimeSpan ReadBatchTimeout(SettingsRootNamespace settingsNamespace, } /// - /// 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. + /// Settles how many messages one write handles. The default splits the transport's concurrency + /// across the writers, so that each can fill a batch while the others are writing. + /// is the resolved writer count, see . /// - public static int ResolveBatchSize(int? configured, int transportMaxConcurrency) => - configured ?? Math.Min(DefaultBatchSize, transportMaxConcurrency); + public static int ResolveBatchSize(int? configured, int transportMaxConcurrency, int writers) => + configured ?? Math.Max(1, Math.Min(transportMaxConcurrency, transportMaxConcurrency / writers)); /// /// Settles how many writers a pipeline actually gets. A storage whose batches are not safe to @@ -107,11 +108,6 @@ 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; diff --git a/src/ServiceControl.Transports.ASBS.Tests/ServiceControlAuditEndpointTests.cs b/src/ServiceControl.Transports.ASBS.Tests/ServiceControlAuditEndpointTests.cs index 78da3c372c..20c547e296 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 78da3c372c..20c547e296 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 453e78e736..f70a724658 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 a0a73cc855..56579e59cd 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 78da3c372c..20c547e296 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 78da3c372c..197afd5970 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs b/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs index 7abb2b9b47..063c32b3e4 100644 --- a/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs +++ b/src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs @@ -15,11 +15,15 @@ public class PostgreSqlTransportCustomization() : TransportCustomization 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 78da3c372c..20c547e296 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 78da3c372c..20c547e296 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 78da3c372c..20c547e296 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 78da3c372c..20c547e296 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 78da3c372c..197afd5970 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ No newline at end of file diff --git a/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs b/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs index c873ca6379..35be6b3a11 100644 --- a/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs +++ b/src/ServiceControl.Transports.SqlServer/SqlServerTransportCustomization.cs @@ -20,12 +20,16 @@ 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 78da3c372c..20c547e296 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 32; } \ 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 61355b073a..465bba2734 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() => 100; + private static partial int GetTransportDefaultConcurrency() => 10; } \ No newline at end of file diff --git a/src/ServiceControl.Transports/TransportCustomization.cs b/src/ServiceControl.Transports/TransportCustomization.cs index b2086f989f..f7d4142eba 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 ??= TransportSettings.DefaultIngestionMaxConcurrency; + transportSettings.MaxConcurrency ??= 10; endpointConfiguration.UseTransport(transport); } @@ -93,7 +93,7 @@ public void CustomizeAuditEndpoint(EndpointConfiguration endpointConfiguration, var transport = CreateTransport(transportSettings); CustomizeTransportForAuditEndpoint(endpointConfiguration, transport, transportSettings); - transportSettings.MaxConcurrency ??= TransportSettings.DefaultIngestionMaxConcurrency; + transportSettings.MaxConcurrency ??= 32; endpointConfiguration.SendOnly(); diff --git a/src/ServiceControl.Transports/TransportSettings.cs b/src/ServiceControl.Transports/TransportSettings.cs index a58089a8bf..246879e482 100644 --- a/src/ServiceControl.Transports/TransportSettings.cs +++ b/src/ServiceControl.Transports/TransportSettings.cs @@ -15,11 +15,6 @@ 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 bada2bb8b4..647b549024 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 ??= TransportSettings.DefaultIngestionMaxConcurrency; + transportSettings.MaxConcurrency ??= 10; 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 3fb9ccd03f..c601c07b57 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 is 25, or the transport's concurrency if that is lower. + /// The most messages one write handles. Null splits the transport's concurrency across the writers. /// public int? ErrorIngestionBatchSize { get; set; } diff --git a/src/ServiceControl/Operations/ErrorIngestion.cs b/src/ServiceControl/Operations/ErrorIngestion.cs index 76ad0bbc03..6bf57e1e38 100644 --- a/src/ServiceControl/Operations/ErrorIngestion.cs +++ b/src/ServiceControl/Operations/ErrorIngestion.cs @@ -45,13 +45,14 @@ public ErrorIngestion( throw new ArgumentException("MaxConcurrency is not set in TransportSettings"); } - MaxBatchSize = IngestionSettingsReader.ResolveBatchSize(settings.ErrorIngestionBatchSize, transportSettings.MaxConcurrency.Value); + var writers = IngestionSettingsReader.ResolveMaxParallelWriters(settings.ErrorIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.ErrorIngestionMaxParallelWriters), logger); + MaxBatchSize = IngestionSettingsReader.ResolveBatchSize(settings.ErrorIngestionBatchSize, transportSettings.MaxConcurrency.Value, writers); pipeline = new IngestionPipeline( new IngestionPipelineSettings { BatchSize = MaxBatchSize, - MaxWriters = IngestionSettingsReader.ResolveMaxParallelWriters(settings.ErrorIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.ErrorIngestionMaxParallelWriters), logger), + MaxWriters = writers, BatchTimeout = settings.ErrorIngestionBatchTimeout }, IngestBatch, From 38778bd21f4ab80437eda921df9082a34ee26ab8 Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Wed, 21 Oct 2026 12:34:19 +0800 Subject: [PATCH 3/4] Change default write parallelism to 2 to be a factor of the default transport parallelism --- docs/ingestion-pipeline.md | 4 ++-- .../Ingestion/IngestionSettingsReaderTests.cs | 2 +- .../Ingestion/IngestionSettingsReader.cs | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/ingestion-pipeline.md b/docs/ingestion-pipeline.md index 71ba862e82..2b3d2e358a 100644 --- a/docs/ingestion-pipeline.md +++ b/docs/ingestion-pipeline.md @@ -47,7 +47,7 @@ Each instance reads its settings from its own settings root | Setting | Default | Range | What it does | | --- | --- | --- | --- | | `ErrorIngestionBatchSize` / `AuditIngestionBatchSize` | the transport's `MaximumConcurrencyLevel` divided by the writers, at least 1 | 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 | +| `ErrorIngestionMaxParallelWriters` / `AuditIngestionMaxParallelWriters` | `2` 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: @@ -95,7 +95,7 @@ someone actually asked for rather than a default. | Audit, SQL Server and PostgreSQL | yes | The audit unit of work factory supports concurrent batches. | | Audit, in memory | no | Test storage only. | -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. +Error ingestion defaults to two writers on SQL Server and PostgreSQL and one on RavenDB. Audit ingestion defaults to two writers on all three providers. In-memory test storage uses one writer. ### The ordering that concurrency does not excuse diff --git a/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs index e566097c44..a352cf0c64 100644 --- a/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs +++ b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs @@ -114,7 +114,7 @@ public void An_out_of_range_value_is_taken_as_it_stands_when_validation_is_off() public void A_storage_that_takes_concurrent_batches_gets_the_default_when_nothing_is_configured() => Assert.That( IngestionSettingsReader.ResolveMaxParallelWriters(null, storageSupportsConcurrentBatches: true, SettingName, NullLogger.Instance), - Is.EqualTo(IngestionSettingsReader.DefaultMaxParallelWriters)); + Is.EqualTo(2)); [Test] public void A_storage_that_takes_concurrent_batches_gets_what_is_configured() => diff --git a/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs b/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs index 0ced632aab..1410fdfeae 100644 --- a/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs +++ b/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs @@ -106,7 +106,7 @@ static Exception Invalid(string message) /// /// What a storage whose batches are safe to interleave gets when nothing is configured. /// - public const int DefaultMaxParallelWriters = 4; + public const int DefaultMaxParallelWriters = 2; public static readonly TimeSpan DefaultBatchTimeout = TimeSpan.FromMilliseconds(100); From b21c2b6be8a20ba669cd0fd7f412f3dc68aa43d6 Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Wed, 21 Oct 2026 12:41:39 +0800 Subject: [PATCH 4/4] Update approvals --- .../APIApprovals.PlatformSampleSettings.approved.txt | 2 +- .../APIApprovals.PlatformSampleSettings.approved.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt index 44ef85404e..f0948c181f 100644 --- a/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt +++ b/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt @@ -64,7 +64,7 @@ "TimeToRestartAuditIngestionAfterFailure": "00:01:00", "AuditIngestionBatchSize": null, "AuditIngestionMaxParallelWriters": null, - "AuditIngestionBatchTimeout": "00:00:00", + "AuditIngestionBatchTimeout": "00:00:00.1000000", "EnableFullTextSearchOnBodies": true, "ShutdownTimeout": "00:00:05" } \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt index abd4c98313..b7303ff3a8 100644 --- a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt +++ b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt @@ -79,7 +79,7 @@ "TimeToRestartErrorIngestionAfterFailure": "00:01:00", "ErrorIngestionBatchSize": null, "ErrorIngestionMaxParallelWriters": null, - "ErrorIngestionBatchTimeout": "00:00:00", + "ErrorIngestionBatchTimeout": "00:00:00.1000000", "MaximumConcurrencyLevel": null, "RetryHistoryDepth": 10, "RemoteInstances": [],