Driver config reporting — stage 2: full DRIVER_CONFIG report

src/Cassandra.IntegrationTests/Core/StartupOptionsTests.cs

@@ -15,6 +15,7 @@
 //
 
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 
@@ -68,16 +69,126 @@ public async Task Should_ReportDriverConfig_OnlyOnTheControlConnection()
 
             _cluster.Connect();
 
+            // That exactly one connection reports the option, and that a pool connection was there not to, is
+            // asserted by the helper.
+            var report = await GetDriverConfigReportAsync().ConfigureAwait(false);
+
+            Assert.AreEqual(DriverConfigReporter.SchemaVersion, report["version"].Value<int>());
+        }
+
+        [Test]
+        public async Task Should_ReportTheEffectiveConfiguration_On_TheControlConnection()
+        {
+            _simulacronCluster = await SimulacronCluster.CreateNewAsync(1).ConfigureAwait(false);
+            _cluster = BuildClusterBuilder()
+                       .WithLoadBalancingPolicy(new TokenAwarePolicy(new DCAwareRoundRobinPolicy("dc1")))
+                       .WithReconnectionPolicy(new ConstantReconnectionPolicy(500))
+                       .WithRetryPolicy(FallthroughRetryPolicy.Instance)
+                       .WithSpeculativeExecutionPolicy(new ConstantSpeculativeExecutionPolicy(100, 2))
+                       .WithQueryOptions(new QueryOptions().SetConsistencyLevel(ConsistencyLevel.LocalQuorum).SetPageSize(1234))
+                       .WithSocketOptions(new SocketOptions().SetConnectTimeoutMillis(3000).SetReadTimeoutMillis(7000))
+                       .Build();
+
+            _cluster.Connect();
+
+            var report = await GetDriverConfigReportAsync().ConfigureAwait(false);
+
+            // Every configured setting arrives over a real connection, in the shape the schema prescribes. The
+            // full set of groups and the conformance of the document itself are covered by the unit tests; this
+            // asserts that what the builder was given is what the server is told.
+            var connection = report["connection"];
+            Assert.AreEqual(3000, connection["connect"]["timeout-ms"].Value<int>());
+            Assert.AreEqual(7000, connection["read"]["timeout-ms"].Value<int>());
+            Assert.AreEqual(PoolingOptions.DefaultMaxRequestsPerConnection, connection["requests"]["in-flight"]["max"].Value<int>());
+            Assert.AreEqual("constant", connection["reconnection"]["policy"]["type"].Value<string>());
+            Assert.AreEqual(500, connection["reconnection"]["policy"]["delay-ms"].Value<int>());
+            // The group is absent rather than carrying a flag when TLS is off.
+            Assert.IsNull(connection["tls"]);
+
+            var query = report["query"];
+            Assert.AreEqual("fallthrough", query["retry"]["policy"]["type"].Value<string>());
+            Assert.AreEqual("constant", query["speculative-execution"]["policy"]["type"].Value<string>());
+            Assert.AreEqual(2, query["speculative-execution"]["policy"]["max-executions"].Value<int>());
+            Assert.AreEqual("token-aware", query["load-balancing"]["policy"]["type"].Value<string>());
+            Assert.AreEqual("shuffle", query["load-balancing"]["policy"]["load-distribution"].Value<string>());
+            Assert.AreEqual("dc", query["load-balancing"]["node-preference"]["type"].Value<string>());
+            Assert.AreEqual("dc1", query["load-balancing"]["node-preference"]["local-dc"].Value<string>());
+            Assert.AreEqual("LOCAL_QUORUM", query["defaults"]["consistency"].Value<string>());
+            Assert.AreEqual(1234, query["defaults"]["page"]["size"].Value<int>());
+        }
+
+        [Test]
+        public async Task Should_ReportAnInferredDatacenter_When_NoneIsConfigured()
+        {
+            _simulacronCluster = await SimulacronCluster.CreateNewAsync(1).ConfigureAwait(false);
+            _cluster = BuildClusterBuilder().Build();
+
+            _cluster.Connect();
+
+            var report = await GetDriverConfigReportAsync().ConfigureAwait(false);
+
+            // The default policy chain infers the datacenter from the node the control connection uses, which is
+            // not known while the report is being built, so only the preference itself is reported.
+            var preference = report["query"]["load-balancing"]["node-preference"];
+            Assert.AreEqual("dc-auto", preference["type"].Value<string>());
+            Assert.IsNull(preference["local-dc"]);
+        }
+
+        /// <summary>
+        /// The startup options every connection sent, each paired with the connection that sent it, having checked
+        /// that a pool connection opened alongside the control connection.
+        /// </summary>
+        /// <remarks>
+        /// Kept paired rather than flattened to a list of options because every claim these tests make about
+        /// <c>DRIVER_CONFIG</c> is a claim about <em>which</em> connections carry it, so the other connections have
+        /// to be visible to be asserted about. Counting startup messages instead would also be satisfied by a run
+        /// where only the control connection had opened, or where one connection sent two of them, which is why the
+        /// number of distinct connections is what gets checked.
+        /// </remarks>
+        private async Task<IList<StartupOnConnection>> GetStartupsByConnectionAsync()
+        {
             var startupLogs = await _simulacronCluster.GetQueriesAsync(null, QueryType.Startup).ConfigureAwait(false);
-            var startupMessages = startupLogs.Select(log => log.Frame.GetStartupMessage()).ToList();
+            var startups = startupLogs
+                           .Select(log => new StartupOnConnection(log.Connection, log.Frame.GetStartupMessage()))
+                           .ToList();
 
-            Assert.GreaterOrEqual(startupMessages.Count, 2, "Expected at least the control connection and one pool connection");
+            Assert.GreaterOrEqual(
+                startups.Select(startup => startup.Connection).Distinct().Count(),
+                2,
+                "Expected at least the control connection and one pool connection");
 
-            var driverConfigMessages = startupMessages.Where(m => m.ContainsKey(DriverConfigReporter.DriverConfigOption)).ToList();
-            Assert.AreEqual(1, driverConfigMessages.Count, "Only the control connection should report the DRIVER_CONFIG option");
+            return startups;
+        }
 
-            var report = JObject.Parse(driverConfigMessages.Single()[DriverConfigReporter.DriverConfigOption]);
-            Assert.AreEqual(DriverConfigReporter.SchemaVersion, report["version"].Value<int>());
+        /// <summary>
+        /// The single <c>DRIVER_CONFIG</c> report the control connection sent, having checked that no other
+        /// connection sent one. Which connection is the control connection is known by elimination, since
+        /// Simulacron identifies a connection only by its client socket.
+        /// </summary>
+        private async Task<JObject> GetDriverConfigReportAsync()
+        {
+            var startups = await GetStartupsByConnectionAsync().ConfigureAwait(false);
+            var reporting = startups
+                            .Where(startup => startup.Options.ContainsKey(DriverConfigReporter.DriverConfigOption))
+                            .ToList();
+
+            Assert.AreEqual(
+                1, reporting.Count, "Exactly one connection, the control connection, should report the DRIVER_CONFIG option");
+
+            return JObject.Parse(reporting.Single().Options[DriverConfigReporter.DriverConfigOption]);
+        }
+
+        private class StartupOnConnection
+        {
+            public StartupOnConnection(string connection, IDictionary<string, string> options)
+            {
+                Connection = connection;
+                Options = options;
+            }
+
+            public string Connection { get; }
+
+            public IDictionary<string, string> Options { get; }
         }
 
         [Test]
@@ -88,12 +199,10 @@ public async Task Should_NotReportDriverConfig_When_ReportingIsDisabled()
 
             _cluster.Connect();
 
-            var startupLogs = await _simulacronCluster.GetQueriesAsync(null, QueryType.Startup).ConfigureAwait(false);
-            var startupMessages = startupLogs.Select(log => log.Frame.GetStartupMessage()).ToList();
+            var startups = await GetStartupsByConnectionAsync().ConfigureAwait(false);
 
-            Assert.GreaterOrEqual(startupMessages.Count, 2, "Expected at least the control connection and one pool connection");
             Assert.IsTrue(
-                startupMessages.All(m => !m.ContainsKey(DriverConfigReporter.DriverConfigOption)),
+                startups.All(startup => !startup.Options.ContainsKey(DriverConfigReporter.DriverConfigOption)),
                 "No connection should report the DRIVER_CONFIG option when reporting is disabled");
         }
 
@@ -105,11 +214,12 @@ public async Task Should_ReportTheSameSessionId_For_EveryConnectionOfTheSameClus
 
             _cluster.Connect();
 
-            var startupLogs = await _simulacronCluster.GetQueriesAsync(null, QueryType.Startup).ConfigureAwait(false);
-            var sessionIds = startupLogs
-                              .Select(log => log.Frame.GetStartupMessage()[StartupOptionsFactory.SessionIdOption])
-                              .Distinct()
-                              .ToList();
+            // Through the helper, so that "every connection" is checked against connections that actually opened:
+            // a run with only the control connection would satisfy this on its own.
+            var sessionIds = (await GetStartupsByConnectionAsync().ConfigureAwait(false))
+                             .Select(startup => startup.Options[StartupOptionsFactory.SessionIdOption])
+                             .Distinct()
+                             .ToList();
 
             Assert.AreEqual(1, sessionIds.Count, "Every connection of the same Cluster instance should report the same SESSION_ID");
             Assert.IsTrue(Guid.TryParse(sessionIds.Single(), out _), "SESSION_ID should be a valid guid");

src/Cassandra.Tests/BuilderTests.cs

@@ -16,6 +16,7 @@
 
 using System;
 using System.Linq;
+using System.Threading;
 using System.Net;
 using Cassandra.Connections;
 using NUnit.Framework;
@@ -249,6 +250,45 @@ public void Should_ThrowArgumentException_When_ProvidedMaxSchemaAgreementsWaitSe
             Assert.That(ex.Message, Is.EqualTo("Max schema agreement wait must be greater than zero"));
         }
 
+        [Test]
+        [TestCase(0)]
+        [TestCase(-2)]
+        [TestCase(int.MinValue)]
+        public void Should_ThrowArgumentException_When_ProvidedQueryTimeoutIsInvalid(int queryAbortTimeout)
+        {
+            // 0 is not "no timeout": the synchronous paths hand it to Task.Wait, which returns at once, so every
+            // request would time out before completing. Below Timeout.Infinite, Task.Wait throws instead.
+            var builder = Cluster.Builder();
+
+            Assert.Throws<ArgumentException>(() => builder.WithQueryTimeout(queryAbortTimeout));
+        }
+
+        [Test]
+        [TestCase(0)]
+        [TestCase(-2)]
+        public void Should_ThrowArgumentException_When_AConfigurationIsBuiltWithAnInvalidQueryTimeout(int queryAbortTimeout)
+        {
+            // ClientOptions is public and can be handed straight to Configuration, bypassing the builder, so the
+            // check has to hold there too: no cluster may come into existence with a timeout that would make
+            // every request fail.
+            Assert.Throws<ArgumentException>(
+                () => new TestConfigurationBuilder { ClientOptions = new ClientOptions(false, queryAbortTimeout, null) }.Build());
+        }
+
+        [Test]
+        [TestCase(1)]
+        [TestCase(30000)]
+        [TestCase(Timeout.Infinite)]
+        public void Should_AcceptQueryTimeout_When_ItIsPositiveOrInfinite(int queryAbortTimeout)
+        {
+            var config = Cluster.Builder()
+                                .AddContactPoint("192.168.1.10")
+                                .WithQueryTimeout(queryAbortTimeout)
+                                .GetConfiguration();
+
+            Assert.AreEqual(queryAbortTimeout, config.ClientOptions.QueryAbortTimeout);
+        }
+
         [Test]
         public void Should_ReturnCorrectMaxSchemaAgreementsWaitSeconds_When_ValueIsProvidedToBuilder()
         {

src/Cassandra.Tests/Cassandra.Tests.csproj

@@ -16,6 +16,14 @@
   <PropertyGroup Condition="$([System.Text.RegularExpressions.Regex]::IsMatch('$(TargetFramework)', '^net\d$'))">
     <DefineConstants>$(DefineConstants);NETCOREAPP</DefineConstants>
   </PropertyGroup>
+  <!-- JsonSchema.Net reaches System.Text.Json 10 and System.Collections.Immutable 10 through Json.More.Net,
+       whose netstandard2.0 group requires them outright, so they cannot be pinned lower. Those support net8 and
+       above, and warn on anything older, so the validator is only referenced there. The report is one code path
+       with no per-framework behaviour, so validating it on net8/net9 establishes its conformance for every
+       target; only the assertions that need the validator are skipped elsewhere. -->
+  <PropertyGroup Condition="'$(TargetFramework)' == 'net8' Or '$(TargetFramework)' == 'net9'">
+    <DefineConstants>$(DefineConstants);JSON_SCHEMA_VALIDATOR</DefineConstants>
+  </PropertyGroup>
   <ItemGroup>
     <ProjectReference Include="..\Cassandra\Cassandra.csproj">
       <SetTargetFramework>TargetFramework=netstandard2.0</SetTargetFramework>
@@ -33,6 +41,12 @@
     <PackageReference Include="System.Runtime.InteropServices.RuntimeInformation" Version="4.3.0" />
     <PackageReference Include="XunitXml.TestLogger" Version="8.0.0" />
     <PackageReference Include="Newtonsoft.Json" Version="13.0.4" />
+    <!-- Validates the DRIVER_CONFIG report against the normative v1 schema shipped alongside it. Test only, and
+         only where its dependencies support the target framework: see the JSON_SCHEMA_VALIDATOR property above.
+         Pinned to the 8.0.x line, the last one published under a plain MIT license: from 9.0.0 the NuGet binaries
+         carry an Open Source Maintenance Fee EULA that asks revenue-generating users for a monthly fee. -->
+    <PackageReference Include="JsonSchema.Net" Version="8.0.5"
+                      Condition="'$(TargetFramework)' == 'net8' Or '$(TargetFramework)' == 'net9'" />
     <ProjectReference Include="..\Extensions\Cassandra.OpenTelemetry\Cassandra.OpenTelemetry.csproj">
       <SetTargetFramework>TargetFramework=netstandard2.0</SetTargetFramework>
     </ProjectReference>
@@ -41,6 +55,11 @@
     <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
     <PackageReference Include="System.Threading.Tasks.Parallel" Version="4.3.0" />
   </ItemGroup>
+  <ItemGroup>
+    <!-- The normative DRIVER_CONFIG v1 schema, shipped verbatim so the reporter is checked against the real
+         thing rather than against a restatement of it. -->
+    <EmbeddedResource Include="Requests\driver-config-report-v1.schema.json" />
+  </ItemGroup>
   <ItemGroup>
     <Service Include="{82a7f48d-3b50-4b1e-b82e-3ada8210c358}" />
   </ItemGroup>

src/Cassandra.Tests/ExecutionProfiles/RequestOptionsTests.cs

@@ -0,0 +1,65 @@
+//
+//      Copyright (C) ScyllaDB
+//
+//   Licensed under the Apache License, Version 2.0 (the "License");
+//   you may not use this file except in compliance with the License.
+//   You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//   Unless required by applicable law or agreed to in writing, software
+//   distributed under the License is distributed on an "AS IS" BASIS,
+//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//   See the License for the specific language governing permissions and
+//   limitations under the License.
+//
+
+using System;
+using System.Threading;
+
+using NUnit.Framework;
+using Assert = NUnit.Framework.Legacy.ClassicAssert;
+
+namespace Cassandra.Tests.ExecutionProfiles
+{
+    [TestFixture]
+    public class RequestOptionsTests
+    {
+        [Test]
+        public void Should_ScaleTheQueryAbortTimeout_By_TheAmountOfQueries()
+        {
+            var requestOptions = RequestOptionsTests.RequestOptionsWithQueryTimeout(1000);
+
+            Assert.AreEqual(1000, requestOptions.GetQueryAbortTimeout(1));
+            Assert.AreEqual(2000, requestOptions.GetQueryAbortTimeout(2));
+        }
+
+        [Test]
+        public void Should_PreserveTheInfiniteSentinel_When_ScalingTheQueryAbortTimeout()
+        {
+            // Scaling has to leave Timeout.Infinite alone: multiplying it would yield -2 or lower, which
+            // Task.Wait rejects outright, so callers such as Metadata.RefreshSchema would throw for a cluster
+            // configured to wait indefinitely rather than wait.
+            var requestOptions = RequestOptionsTests.RequestOptionsWithQueryTimeout(Timeout.Infinite);
+
+            Assert.AreEqual(Timeout.Infinite, requestOptions.GetQueryAbortTimeout(1));
+            Assert.AreEqual(Timeout.Infinite, requestOptions.GetQueryAbortTimeout(2));
+        }
+
+        [Test]
+        public void Should_ThrowArgumentException_When_TheAmountOfQueriesIsNotPositive()
+        {
+            var requestOptions = RequestOptionsTests.RequestOptionsWithQueryTimeout(1000);
+
+            Assert.Throws<ArgumentException>(() => requestOptions.GetQueryAbortTimeout(0));
+        }
+
+        private static Cassandra.ExecutionProfiles.IRequestOptions RequestOptionsWithQueryTimeout(int queryAbortTimeout)
+        {
+            return new TestConfigurationBuilder
+            {
+                ClientOptions = new ClientOptions(false, queryAbortTimeout, null)
+            }.Build().DefaultRequestOptions;
+        }
+    }
+}