Use SQL system catalog views to check for the presence of a Recoverab…

…le column. This removes the need for SELECT permissions to send a message to a queue table. (#1452) Co-authored-by: Marc Wils <marc@marcwils.be.eu.org>
Particular · Oct 17, 2024 · 7ec3284 · 7ec3284
1 parent c6b5359
commit 7ec3284
Show file tree

Hide file tree

Showing 11 changed files with 45 additions and 39 deletions.
diff --git a/...er.AcceptanceTests/NativeTimeouts/When_configured_to_purge_expired_messages_at_startup.cs b/...er.AcceptanceTests/NativeTimeouts/When_configured_to_purge_expired_messages_at_startup.cs
@@ -1,4 +1,4 @@
-namespace NServiceBus.Transport.SqlServer.AcceptanceTests.NativeTimeouts
+namespace NServiceBus.Transport.SqlServer.AcceptanceTests.NativeTimeouts
 {
     using System;
     using System.Collections.Generic;

diff --git a/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_checking_schema.cs b/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_checking_schema.cs
@@ -1,4 +1,4 @@
-namespace NServiceBus.Transport.SqlServer.IntegrationTests
+namespace NServiceBus.Transport.SqlServer.IntegrationTests
 {
     using System;
     using System.Threading;
@@ -23,7 +23,7 @@ public async Task SetUp()
 
             await ResetQueue(addressParser, sqlConnectionFactory);
 
-            queue = new TableBasedQueue(addressParser.Parse(QueueTableName).QualifiedTableName, QueueTableName, false);
+            queue = new TableBasedQueue(addressParser.Parse(QueueTableName), QueueTableName, false);
         }
 
         [Test]

diff --git a/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_dispatching_messages.cs b/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_dispatching_messages.cs
@@ -1,4 +1,4 @@
-namespace NServiceBus.Transport.SqlServer.IntegrationTests
+namespace NServiceBus.Transport.SqlServer.IntegrationTests
 {
     using System;
     using System.Collections.Generic;
@@ -112,8 +112,7 @@ async Task PrepareAsync(CancellationToken cancellationToken = default)
         Task PurgeOutputQueue(QueueAddressTranslator addressTranslator, CancellationToken cancellationToken = default)
         {
             purger = new QueuePurger(sqlConnectionFactory);
-            var queueAddress = addressTranslator.Parse(ValidAddress).QualifiedTableName;
-            queue = new TableBasedQueue(queueAddress, ValidAddress, true);
+            queue = new TableBasedQueue(addressTranslator.Parse(ValidAddress), ValidAddress, true);
 
             return purger.Purge(queue, cancellationToken);
         }

diff --git a/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_message_receive_takes_long.cs b/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_message_receive_takes_long.cs
@@ -1,4 +1,4 @@
-namespace NServiceBus.Transport.SqlServer.IntegrationTests
+namespace NServiceBus.Transport.SqlServer.IntegrationTests
 {
     using System;
     using System.Collections.Generic;
@@ -28,7 +28,7 @@ public async Task SetUp()
 
             await CreateQueueIfNotExists(addressParser, sqlConnectionFactory);
 
-            queue = new TableBasedQueue(addressParser.Parse(QueueTableName).QualifiedTableName, QueueTableName, true);
+            queue = new TableBasedQueue(addressParser.Parse(QueueTableName), QueueTableName, true);
         }
 
         [Test]

diff --git a/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_receiving_messages.cs b/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_receiving_messages.cs
@@ -1,4 +1,4 @@
-namespace NServiceBus.Transport.SqlServer.IntegrationTests
+namespace NServiceBus.Transport.SqlServer.IntegrationTests
 {
     using System;
     using System.Linq;
@@ -23,7 +23,7 @@ public async Task Should_stop_receiving_messages_after_first_unsuccessful_receiv
 
             var parser = new QueueAddressTranslator("nservicebus", "dbo", null, null);
             var inputQueueName = "input";
-            var inputQueueAddress = parser.Parse(inputQueueName).Address;
+            var inputQueueAddress = parser.Parse(inputQueueName);
             var inputQueue = new FakeTableBasedQueue(inputQueueAddress, queueSize, successfulReceives);
 
             var connectionString = Environment.GetEnvironmentVariable("SqlServerTransportConnectionString") ?? @"Data Source=.\SQLEXPRESS;Initial Catalog=nservicebus;Integrated Security=True";
@@ -35,7 +35,7 @@ public async Task Should_stop_receiving_messages_after_first_unsuccessful_receiv
             };
 
             transport.Testing.QueueFactoryOverride = qa =>
-                qa == inputQueueAddress ? inputQueue : new TableBasedQueue(parser.Parse(qa).QualifiedTableName, qa, true);
+                qa == inputQueueAddress.Address ? inputQueue : new TableBasedQueue(parser.Parse(qa), qa, true);
 
             var receiveSettings = new ReceiveSettings("receiver", new Transport.QueueAddress(inputQueueName), true, false, "error");
             var hostSettings = new HostSettings("IntegrationTests", string.Empty, new StartupDiagnosticEntries(),
@@ -87,7 +87,7 @@ class FakeTableBasedQueue : TableBasedQueue
             int queueSize;
             int successfulReceives;
 
-            public FakeTableBasedQueue(string address, int queueSize, int successfulReceives) : base(address, "", true)
+            public FakeTableBasedQueue(CanonicalQueueAddress address, int queueSize, int successfulReceives) : base(address, "", true)
             {
                 this.queueSize = queueSize;
                 this.successfulReceives = successfulReceives;

diff --git a/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_recoverable_column_is_removed.cs b/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_recoverable_column_is_removed.cs
@@ -1,4 +1,4 @@
-namespace NServiceBus.Transport.SqlServer.IntegrationTests
+namespace NServiceBus.Transport.SqlServer.IntegrationTests
 {
     using System;
     using System.Collections.Generic;

diff --git a/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_using_ttbr.cs b/src/NServiceBus.Transport.SqlServer.IntegrationTests/When_using_ttbr.cs
@@ -1,4 +1,4 @@
-namespace NServiceBus.Transport.SqlServer.IntegrationTests
+namespace NServiceBus.Transport.SqlServer.IntegrationTests
 {
     using System;
     using System.Collections.Generic;
@@ -134,7 +134,7 @@ Task PurgeOutputQueue(QueueAddressTranslator addressParser, CancellationToken ca
         {
             purger = new QueuePurger(sqlConnectionFactory);
             var queueAddress = addressParser.Parse(ValidAddress);
-            queue = new TableBasedQueue(queueAddress.QualifiedTableName, queueAddress.Address, true);
+            queue = new TableBasedQueue(queueAddress, queueAddress.Address, true);
 
             return purger.Purge(queue, cancellationToken);
         }

diff --git a/src/NServiceBus.Transport.SqlServer/Queuing/SqlConstants.cs b/src/NServiceBus.Transport.SqlServer/Queuing/SqlConstants.cs
@@ -48,7 +48,11 @@ THEN DATEADD(ms, @TimeToBeReceivedMs, GETUTCDATE()) END,
 IF (@NOCOUNT = 'ON') SET NOCOUNT ON;
 IF (@NOCOUNT = 'OFF') SET NOCOUNT OFF;";
 
-        public static readonly string CheckIfTableHasRecoverableText = "SELECT TOP (0) * FROM {0} WITH (NOLOCK);";
+        public static string CheckIfTableHasRecoverableText { get; set; } = @"
+SELECT COUNT(*)
+FROM {0}.sys.columns c
+WHERE c.object_id = OBJECT_ID(N'{1}')
+    AND c.name = 'Recoverable'";
 
         public static readonly string StoreDelayedMessageText =
 @"

diff --git a/src/NServiceBus.Transport.SqlServer/Queuing/TableBasedQueue.cs b/src/NServiceBus.Transport.SqlServer/Queuing/TableBasedQueue.cs
@@ -17,16 +17,17 @@ class TableBasedQueue
     {
         public string Name { get; }
 
-        public TableBasedQueue(string qualifiedTableName, string queueName, bool isStreamSupported)
+        public TableBasedQueue(CanonicalQueueAddress queueAddress, string queueName, bool isStreamSupported)
         {
-            this.qualifiedTableName = qualifiedTableName;
+            qualifiedTableName = queueAddress.QualifiedTableName;
             Name = queueName;
-            receiveCommand = Format(SqlConstants.ReceiveText, this.qualifiedTableName);
-            purgeCommand = Format(SqlConstants.PurgeText, this.qualifiedTableName);
-            purgeExpiredCommand = Format(SqlConstants.PurgeBatchOfExpiredMessagesText, this.qualifiedTableName);
-            checkExpiresIndexCommand = Format(SqlConstants.CheckIfExpiresIndexIsPresent, this.qualifiedTableName);
-            checkNonClusteredRowVersionIndexCommand = Format(SqlConstants.CheckIfNonClusteredRowVersionIndexIsPresent, this.qualifiedTableName);
-            checkHeadersColumnTypeCommand = Format(SqlConstants.CheckHeadersColumnType, this.qualifiedTableName);
+            receiveCommand = Format(SqlConstants.ReceiveText, qualifiedTableName);
+            purgeCommand = Format(SqlConstants.PurgeText, qualifiedTableName);
+            purgeExpiredCommand = Format(SqlConstants.PurgeBatchOfExpiredMessagesText, qualifiedTableName);
+            checkExpiresIndexCommand = Format(SqlConstants.CheckIfExpiresIndexIsPresent, qualifiedTableName);
+            checkNonClusteredRowVersionIndexCommand = Format(SqlConstants.CheckIfNonClusteredRowVersionIndexIsPresent, qualifiedTableName);
+            checkHeadersColumnTypeCommand = Format(SqlConstants.CheckHeadersColumnType, qualifiedTableName);
+            checkRecoverableColumnCommand = Format(SqlConstants.CheckIfTableHasRecoverableText, queueAddress.Catalog, qualifiedTableName);
             this.isStreamSupported = isStreamSupported;
         }
 
@@ -145,23 +146,24 @@ async Task<string> GetSendCommandText(SqlConnection connection, SqlTransaction t
                     return sendCommand;
                 }
 
-                var commandText = Format(SqlConstants.CheckIfTableHasRecoverableText, qualifiedTableName);
-                using (var command = new SqlCommand(commandText, connection, transaction))
+                using (var command = connection.CreateCommand())
                 {
-                    using (var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false))
+                    command.CommandText = checkRecoverableColumnCommand;
+                    command.CommandType = CommandType.Text;
+                    command.Transaction = transaction;
+
+                    var rowsCount = await command.ExecuteScalarAsync<int>(nameof(checkRecoverableColumnCommand), cancellationToken).ConfigureAwait(false);
+                    if (rowsCount > 0)
                     {
-                        for (int fieldIndex = 0; fieldIndex < reader.FieldCount; fieldIndex++)
-                        {
-                            if (string.Equals("Recoverable", reader.GetName(fieldIndex), StringComparison.OrdinalIgnoreCase))
-                            {
-                                cachedSendCommand = Format(SqlConstants.SendTextWithRecoverable, qualifiedTableName);
-                                return cachedSendCommand;
-                            }
-                        }
+                        cachedSendCommand = Format(SqlConstants.SendTextWithRecoverable, qualifiedTableName);
+                        return cachedSendCommand;
                     }
+                    else
+                    {
 
-                    cachedSendCommand = Format(SqlConstants.SendText, qualifiedTableName);
-                    return cachedSendCommand;
+                        cachedSendCommand = Format(SqlConstants.SendText, qualifiedTableName);
+                        return cachedSendCommand;
+                    }
                 }
             }
             finally
@@ -237,6 +239,7 @@ public override string ToString()
         string checkExpiresIndexCommand;
         string checkNonClusteredRowVersionIndexCommand;
         string checkHeadersColumnTypeCommand;
+        string checkRecoverableColumnCommand;
         bool isStreamSupported;
 
         readonly SemaphoreSlim sendCommandLock = new SemaphoreSlim(1, 1);

diff --git a/src/NServiceBus.Transport.SqlServer/Queuing/TableBasedQueueCache.cs b/src/NServiceBus.Transport.SqlServer/Queuing/TableBasedQueueCache.cs
@@ -15,7 +15,7 @@ public TableBasedQueue Get(string destination)
         {
             var address = addressTranslator.Parse(destination);
             var key = Tuple.Create(address.QualifiedTableName, address.Address);
-            var queue = cache.GetOrAdd(key, x => new TableBasedQueue(x.Item1, x.Item2, isStreamSupported));
+            var queue = cache.GetOrAdd(key, x => new TableBasedQueue(address, x.Item2, isStreamSupported));
 
             return queue;
         }

diff --git a/src/NServiceBus.Transport.SqlServer/SqlServerTransportInfrastructure.cs b/src/NServiceBus.Transport.SqlServer/SqlServerTransportInfrastructure.cs
@@ -131,7 +131,7 @@ public async Task ConfigureReceiveInfrastructure(ReceiveSettings[] receiveSettin
 
             var schemaVerification = new SchemaInspector((queue, token) => connectionFactory.OpenNewConnection(token), validateExpiredIndex);
 
-            var queueFactory = transport.Testing.QueueFactoryOverride ?? (queueName => new TableBasedQueue(addressTranslator.Parse(queueName).QualifiedTableName, queueName, !isEncrypted));
+            var queueFactory = transport.Testing.QueueFactoryOverride ?? (queueName => new TableBasedQueue(addressTranslator.Parse(queueName), queueName, !isEncrypted));
 
             //Create delayed delivery infrastructure
             CanonicalQueueAddress delayedQueueCanonicalAddress = null;