Slowing down the polling on database backed queue listeners in the ca…

…se of an apparently idle queue
JasperFx · Jan 8, 2025 · df3bd32 · df3bd32
1 parent 25dfbbc
commit df3bd32
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 7 deletions.
diff --git a/src/Persistence/PersistenceTests/durability_with_local.cs b/src/Persistence/PersistenceTests/durability_with_local.cs
@@ -32,13 +32,14 @@ public async Task should_recover_persisted_messages()
         // Don't use WolverineHost here because you need the existing persisted state!!!!
         using var host2 = await Host.CreateDefaultBuilder().UseWolverine(opts => opts.ConfigureDurableSender(true, false))
             .StartAsync();
-        var counts2 = await host2.Get<IMessageStore>().Admin.FetchCountsAsync();
+        var messageStore = host2.Get<IMessageStore>();
+        var counts2 = await messageStore.Admin.FetchCountsAsync();
 
         var i = 0;
-        while (counts2.Incoming != 1 && i < 10)
+        while (counts2.Incoming != 1 && i < 100)
         {
             await Task.Delay(100.Milliseconds());
-            counts2 = await host2.Get<IMessageStore>().Admin.FetchCountsAsync();
+            counts2 = await messageStore.Admin.FetchCountsAsync();
             i++;
         }
 
@@ -61,7 +62,11 @@ public static void ConfigureDurableSender(this WolverineOptions opts, bool latch
             .ToLocalQueue("one")
             .UseDurableInbox();
 
-        opts.Services.AddMarten(Servers.PostgresConnectionString)
+        opts.Services.AddMarten(opts =>
+            {
+                opts.Connection(Servers.PostgresConnectionString);
+                opts.DisableNpgsqlLogging = true;
+            })
             .IntegrateWithWolverine();
 
         opts.Services.AddSingleton(new ReceivingSettings { Latched = latched });
@@ -85,7 +90,7 @@ public void Handle(ReceivedMessage message, Envelope envelope, ReceivingSettings
 
     public static void Configure(HandlerChain chain)
     {
-        chain.OnException(e => true).Requeue(1000);
+        chain.OnException(e => true).RequeueIndefinitely();
     }
 }
 

diff --git a/src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueueListener.cs b/src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueueListener.cs
@@ -180,11 +180,13 @@ private async Task listenForMessagesAsync()
                 if (messages.Any())
                 {
                     await _receiver.ReceivedAsync(this, messages.ToArray());
+
+                    await Task.Delay(250.Milliseconds());
                 }
                 else
                 {
                     // Slow down if this is a periodically used queue
-                    await Task.Delay(250.Milliseconds());
+                    await Task.Delay(_settings.ScheduledJobPollingTime);
                 }
             }
             catch (Exception e)

diff --git a/src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueueListener.cs b/src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueueListener.cs
@@ -119,7 +119,7 @@ private async Task listenForMessagesAsync()
                 else
                 {
                     // Slow down if this is a periodically used queue
-                    await Task.Delay(250.Milliseconds());
+                    await Task.Delay(_settings.ScheduledJobPollingTime);
                 }
             }
             catch (Exception e)