From 99ef9f52cb6953f1b9ebb759ca7946e5a467606a Mon Sep 17 00:00:00 2001 From: Brandon Ording Date: Thu, 20 Nov 2025 14:48:28 -0500 Subject: [PATCH 1/6] Avoid actually setting unlimited value in policy --- .../BrokerVerifierTests.cs | 3 ++- .../Administration/BrokerVerifier.cs | 4 ++-- .../ManagementApi/Converters/DeliveryLimitConverter.cs | 2 +- .../Administration/ManagementApi/Models/Queue.cs | 5 ++++- 4 files changed, 9 insertions(+), 5 deletions(-) diff --git a/src/NServiceBus.Transport.RabbitMQ.Tests/BrokerVerifierTests.cs b/src/NServiceBus.Transport.RabbitMQ.Tests/BrokerVerifierTests.cs index cbfe321fe..cff4e5a41 100644 --- a/src/NServiceBus.Transport.RabbitMQ.Tests/BrokerVerifierTests.cs +++ b/src/NServiceBus.Transport.RabbitMQ.Tests/BrokerVerifierTests.cs @@ -49,8 +49,9 @@ public async Task ValidateDeliveryLimit_Should_Set_Delivery_Limit_Policy() for (int i = 0; i < attempts; i++) { var queue = await managementClient.GetQueue(queueName); + var deliveryLimit = queue.GetDeliveryLimit(); - if (queue.DeliveryLimit == -1 && queue.AppliedPolicyName == policyName) + if (deliveryLimit == Queue.BigValueInsteadOfActuallyUnlimited && queue.AppliedPolicyName == policyName) { // Policy applied successfully return; diff --git a/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs b/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs index ed3fc43a0..a29c5aa14 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs @@ -171,7 +171,7 @@ bool ShouldOverrideDeliveryLimit(Queue queue) var limit = queue.GetDeliveryLimit(); - if (limit == -1) + if (limit is Queue.UnlimitedDeliveryLimitIndicatorValue or Queue.BigValueInsteadOfActuallyUnlimited) { return false; } @@ -201,7 +201,7 @@ async Task SetDeliveryLimitViaPolicy(Queue queue, CancellationToken cancellation var policy = new Policy { ApplyTo = PolicyTarget.QuorumQueues, - Definition = new PolicyDefinition { DeliveryLimit = -1 }, + Definition = new PolicyDefinition { DeliveryLimit = Queue.BigValueInsteadOfActuallyUnlimited }, Pattern = queue.Name, Priority = 0 }; diff --git a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Converters/DeliveryLimitConverter.cs b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Converters/DeliveryLimitConverter.cs index 951f9f129..71545e97e 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Converters/DeliveryLimitConverter.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Converters/DeliveryLimitConverter.cs @@ -20,7 +20,7 @@ public override int Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSeri if (string.Equals(value, "unlimited", StringComparison.OrdinalIgnoreCase)) { - return -1; + return Queue.UnlimitedDeliveryLimitIndicatorValue; } throw new JsonException($"Unexpected string value for delivery limit: {value}"); diff --git a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs index 04ab08a82..86b106760 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs @@ -67,6 +67,9 @@ public int GetDeliveryLimit() limit = 0; } - return limit ?? -1; + return limit ?? UnlimitedDeliveryLimitIndicatorValue; } + + public const int UnlimitedDeliveryLimitIndicatorValue = -1_000; + public const int BigValueInsteadOfActuallyUnlimited = 100_000; } From 8c90c21193da3531420e89c931d30127ce141940 Mon Sep 17 00:00:00 2001 From: Brandon Ording Date: Thu, 20 Nov 2025 17:20:56 -0500 Subject: [PATCH 2/6] Fix comment typo --- .../Administration/ManagementApi/Models/Queue.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs index 86b106760..681c33c9a 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs @@ -43,8 +43,8 @@ public int GetDeliveryLimit() } // RabbitMQ 3.x - // The broker doesn't tell us what the actual delivery limit is, so we have to figure it out - // We have to find the lowest value from the possible places in can be configured + // The broker doesn't tell us what the actual delivery limit is, so we have to figure it out. + // We have to find the lowest value from the possible places it can be configured. int? limit = null; From c5d849eb87e7854299a5465d589a72b70ec6d5b6 Mon Sep 17 00:00:00 2001 From: Brandon Ording Date: Mon, 24 Nov 2025 15:09:06 -0500 Subject: [PATCH 3/6] Treat unlimited value as needing policy update also --- .../Administration/BrokerVerifier.cs | 2 +- .../Administration/ManagementApi/Models/Queue.cs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs b/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs index a29c5aa14..fdd3dd12e 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs @@ -171,7 +171,7 @@ bool ShouldOverrideDeliveryLimit(Queue queue) var limit = queue.GetDeliveryLimit(); - if (limit is Queue.UnlimitedDeliveryLimitIndicatorValue or Queue.BigValueInsteadOfActuallyUnlimited) + if (limit is Queue.BigValueInsteadOfActuallyUnlimited) { return false; } diff --git a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs index 681c33c9a..797899b4a 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs @@ -67,7 +67,7 @@ public int GetDeliveryLimit() limit = 0; } - return limit ?? UnlimitedDeliveryLimitIndicatorValue; + return limit ?? BigValueInsteadOfActuallyUnlimited; } public const int UnlimitedDeliveryLimitIndicatorValue = -1_000; From 33bbc93b85d12214ab1bae265d9d9b49026c5ccc Mon Sep 17 00:00:00 2001 From: Brandon Ording Date: Mon, 24 Nov 2025 15:11:29 -0500 Subject: [PATCH 4/6] Remove unneeded UnlimitedDeliveryLimitIndicatorValue --- .../ManagementApi/Converters/DeliveryLimitConverter.cs | 2 +- .../Administration/ManagementApi/Models/Queue.cs | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Converters/DeliveryLimitConverter.cs b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Converters/DeliveryLimitConverter.cs index 71545e97e..951f9f129 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Converters/DeliveryLimitConverter.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Converters/DeliveryLimitConverter.cs @@ -20,7 +20,7 @@ public override int Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSeri if (string.Equals(value, "unlimited", StringComparison.OrdinalIgnoreCase)) { - return Queue.UnlimitedDeliveryLimitIndicatorValue; + return -1; } throw new JsonException($"Unexpected string value for delivery limit: {value}"); diff --git a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs index 797899b4a..949120469 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Administration/ManagementApi/Models/Queue.cs @@ -70,6 +70,5 @@ public int GetDeliveryLimit() return limit ?? BigValueInsteadOfActuallyUnlimited; } - public const int UnlimitedDeliveryLimitIndicatorValue = -1_000; public const int BigValueInsteadOfActuallyUnlimited = 100_000; } From 171d4f19532de0d6c1ac10b94e614c9b1e4a0760 Mon Sep 17 00:00:00 2001 From: Brandon Ording Date: Mon, 24 Nov 2025 16:12:28 -0500 Subject: [PATCH 5/6] Add failing test for old policy updating --- .../BrokerVerifierTests.cs | 72 ++++++++++++++++++- 1 file changed, 71 insertions(+), 1 deletion(-) diff --git a/src/NServiceBus.Transport.RabbitMQ.Tests/BrokerVerifierTests.cs b/src/NServiceBus.Transport.RabbitMQ.Tests/BrokerVerifierTests.cs index cff4e5a41..5514f0080 100644 --- a/src/NServiceBus.Transport.RabbitMQ.Tests/BrokerVerifierTests.cs +++ b/src/NServiceBus.Transport.RabbitMQ.Tests/BrokerVerifierTests.cs @@ -63,6 +63,76 @@ public async Task ValidateDeliveryLimit_Should_Set_Delivery_Limit_Policy() Assert.Fail($"Policy '{policyName}' was not applied to queue '{queueName}'."); } + [Test] + public async Task ValidateDeliveryLimit_Should_Update_Old_Unlimited_Policy_Created_By_Transport() + { + var managementClient = new ManagementClient(connectionConfiguration); + using var brokerVerifier = new BrokerVerifier(managementClient, BrokerRequirementChecks.None, true); + await brokerVerifier.Initialize(); + + if (brokerVerifier.BrokerVersion < BrokerVerifier.BrokerVersion4) + { + Assert.Ignore("Test not valid for broker versions before 4.0.0"); + } + + var queueName = nameof(ValidateDeliveryLimit_Should_Set_Delivery_Limit_Policy); + var policyName = $"nsb.{queueName}.delivery-limit"; + + var oldUnlimitedPolicy = new Policy + { + ApplyTo = PolicyTarget.QuorumQueues, + Definition = new PolicyDefinition { DeliveryLimit = -1 }, + Pattern = queueName, + Priority = 0 + }; + + await CreateQueue(queueName); + await managementClient.CreatePolicy(policyName, oldUnlimitedPolicy); + + // It can take some time for updated policies to be applied, so we need to wait. + // If this test is randomly failing, consider increasing the attempts + var attempts = 20; + + int deliveryLimit = 0; + + for (int i = 0; i < attempts; i++) + { + var queue = await managementClient.GetQueue(queueName); + deliveryLimit = queue.GetDeliveryLimit(); + + if (deliveryLimit == -1 && queue.AppliedPolicyName == policyName) + { + // Policy applied successfully + break; + } + + await Task.Delay(TimeSpan.FromSeconds(3)); + } + + if (deliveryLimit != -1) + { + Assert.Fail($"Old unlimited Policy '{policyName}' was not applied to queue '{queueName}'."); + } + + await brokerVerifier.ValidateDeliveryLimit(queueName); + + for (int i = 0; i < attempts; i++) + { + var queue = await managementClient.GetQueue(queueName); + deliveryLimit = queue.GetDeliveryLimit(); + + if (deliveryLimit == Queue.BigValueInsteadOfActuallyUnlimited && queue.AppliedPolicyName == policyName) + { + // Policy applied successfully + return; + } + + await Task.Delay(TimeSpan.FromSeconds(3)); + } + + Assert.Fail($"Policy '{policyName}' was not applied to queue '{queueName}'."); + } + [Test] public async Task ValidateDeliveryLimit_Should_Throw_When_Queue_Does_Not_Exist() { @@ -125,7 +195,7 @@ public async Task ValidateDeliveryLimit_Should_Throw_When_A_Policy_On_Queue_Has_ await managementClient.CreatePolicy(policyName, policy).ConfigureAwait(false); - // If this test appears flaky, the delay should be increased to give the broker more time to apply the policy before calling ValidateDeliveryLimit + // If this test appears flaky, the delay should be increased to give the broker more time to apply the oldUnlimitedPolicy before calling ValidateDeliveryLimit await Task.Delay(TimeSpan.FromSeconds(30)); var exception = Assert.ThrowsAsync(async () => await brokerVerifier.ValidateDeliveryLimit(queueName)); From 07b3f59f0cfe9dc4afd513e22e36617d58888371 Mon Sep 17 00:00:00 2001 From: Brandon Ording Date: Mon, 24 Nov 2025 16:14:17 -0500 Subject: [PATCH 6/6] Ensure old policy can be updated --- .../Administration/BrokerVerifier.cs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs b/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs index fdd3dd12e..e47b2d4ac 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Administration/BrokerVerifier.cs @@ -176,6 +176,11 @@ bool ShouldOverrideDeliveryLimit(Queue queue) return false; } + if (limit == -1 && queue.AppliedPolicyName == $"nsb.{queue.Name}.delivery-limit") + { + return true; + } + if (queue.Arguments.DeliveryLimit.HasValue || (queue.EffectivePolicyDefinition?.DeliveryLimit.HasValue ?? false)) { throw new InvalidOperationException($"The delivery limit for '{queue.Name}' is set to the non-default value of '{limit}'. Remove any delivery limit settings from queue arguments, user policies or operator policies to correct this."); @@ -191,13 +196,13 @@ async Task SetDeliveryLimitViaPolicy(Queue queue, CancellationToken cancellation throw new InvalidOperationException($"Cannot create unlimited delivery limit policies in RabbitMQ versions prior to 4.0. The version is: {brokerVersion}."); } - if (!string.IsNullOrEmpty(queue.AppliedPolicyName)) + var policyName = $"nsb.{queue.Name}.delivery-limit"; + + if (!string.IsNullOrEmpty(queue.AppliedPolicyName) && queue.AppliedPolicyName != policyName) { throw new InvalidOperationException($"An unlimited delivery limit policy cannot be applied to the '{queue.Name}' queue because it already has a '{queue.AppliedPolicyName}' policy applied."); } - var policyName = $"nsb.{queue.Name}.delivery-limit"; - var policy = new Policy { ApplyTo = PolicyTarget.QuorumQueues,