Use the testing tool’s ingestion tuning as defaults

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
 

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<IHostedService>().ToArray());
-                Assert.That(host.Services.GetRequiredService<TransportSettings>().MaxConcurrency, Is.EqualTo(10));
+                Assert.That(host.Services.GetRequiredService<TransportSettings>().MaxConcurrency, Is.EqualTo(100));
             }
             finally
             {

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -45,7 +45,7 @@ ILogger<AuditIngestion> 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

src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs

@@ -189,7 +189,7 @@ public int MaxBodySizeToStore
         public TimeSpan TimeToRestartAuditIngestionAfterFailure { get; set; }
 
         /// <summary>
-        /// 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.
         /// </summary>
         public int? AuditIngestionBatchSize { get; set; }
 
@@ -200,7 +200,7 @@ public int MaxBodySizeToStore
         public int? AuditIngestionMaxParallelWriters { get; set; }
 
         /// <summary>
-        /// 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.
         /// </summary>
         public TimeSpan AuditIngestionBatchTimeout { get; set; }
 

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()
     {

src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs

@@ -13,8 +13,7 @@ namespace ServiceControl.Infrastructure.Ingestion;
 public static class IngestionSettingsReader
 {
     /// <summary>
-    /// 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 <see cref="ResolveBatchSize" />.
     /// </summary>
     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);
 
     /// <summary>
-    /// 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
+    /// <see cref="DefaultBatchTimeout" />; an explicit zero does not wait at all.
     /// </summary>
     public static TimeSpan ReadBatchTimeout(SettingsRootNamespace settingsNamespace, string name, bool validateConfiguration)
     {
         if (!SettingsReader.TryRead<string>(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;
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public static int ResolveBatchSize(int? configured, int transportMaxConcurrency) =>
+        configured ?? Math.Min(DefaultBatchSize, transportMaxConcurrency);
+
     /// <summary>
     /// 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)
     /// </summary>
     public const int DefaultMaxParallelWriters = 4;
 
+    /// <summary>
+    /// The most messages one write handles when nothing is configured and the transport allows it.
+    /// </summary>
+    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);

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

src/ServiceControl.Transports.PostgreSql/PostgreSqlTransportCustomization.cs

@@ -15,15 +15,11 @@ public class PostgreSqlTransportCustomization() : TransportCustomization<Postgre
     protected override void CustomizeTransportForPrimaryEndpoint(EndpointConfiguration endpointConfiguration, PostgreSqlTransport transportDefinition, TransportSettings transportSettings)
     {
         transportDefinition.TransportTransactionMode = TransportTransactionMode.SendsAtomicWithReceive;
-
-        transportSettings.MaxConcurrency ??= 10;
     }
 
     protected override void CustomizeTransportForAuditEndpoint(EndpointConfiguration endpointConfiguration, PostgreSqlTransport transportDefinition, TransportSettings transportSettings)
     {
         transportDefinition.TransportTransactionMode = TransportTransactionMode.ReceiveOnly;
-
-        transportSettings.MaxConcurrency ??= 10;
     }
 
     protected override void CustomizeTransportForMonitoringEndpoint(EndpointConfiguration endpointConfiguration, PostgreSqlTransport transportDefinition, TransportSettings transportSettings)

src/ServiceControl.Transports.RabbitMQClassicConventionalRouting.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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }

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;
 }