update sql/postgre transports to latest and tweak ql query (#4549)

Particular · Oct 23, 2024 · 492d1c8 · 492d1c8
1 parent 224aaaf
commit 492d1c8
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 6 deletions.
diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props
@@ -47,8 +47,8 @@
     <PackageVersion Include="NServiceBus.Transport.AzureServiceBus" Version="4.2.3" />
     <PackageVersion Include="NServiceBus.Transport.AzureStorageQueues" Version="13.0.1" />
     <PackageVersion Include="NServiceBus.Transport.Msmq.Sources" Version="3.0.1" />
-    <PackageVersion Include="NServiceBus.Transport.PostgreSql" Version="8.1.4" />
-    <PackageVersion Include="NServiceBus.Transport.SqlServer" Version="8.1.4" />
+    <PackageVersion Include="NServiceBus.Transport.SqlServer" Version="8.1.5" />
+    <PackageVersion Include="NServiceBus.Transport.PostgreSql" Version="8.1.5" />
     <PackageVersion Include="NuGet.Versioning" Version="6.11.1" />
     <PackageVersion Include="NUnit" Version="4.2.2" />
     <PackageVersion Include="NUnit.Analyzers" Version="4.3.0" />

diff --git a/src/ServiceControl.Transports.PostgreSql/PostgreSqlTable.cs b/src/ServiceControl.Transports.PostgreSql/PostgreSqlTable.cs
@@ -1,4 +1,4 @@
-#nullable enable
+#nullable enable
 namespace ServiceControl.Transports.PostgreSql;
 
 class PostgreSqlTable
@@ -7,13 +7,14 @@ public PostgreSqlTable(string name, string schema)
     {
         //HINT: The query approximates queue length value based on max and min of the table sequence.
         fullTableName = $"\"{schema}\".\"{name}\"";
-        //NOTE: Postgres doesn't have NOLOCK since it utilises snapshot isolation by default
+        //      As a result, we want to skip rows locked by other transactions with SKIP LOCKED query hint.
         LengthQuery = $$"""
                         SELECT CASE WHEN (EXISTS (SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '{{schema}}' AND TABLE_NAME = '{{name}}')) THEN
-                          COALESCE(cast(max(seq) - min(seq) + 1 AS int), 0)
+                          COALESCE(cast((SELECT seq FROM {{schema}}."{{name}}" ORDER BY seq DESC LIMIT 1 FOR UPDATE SKIP LOCKED)
+                                      - (SELECT seq FROM {{schema}}."{{name}}" ORDER BY seq ASC  LIMIT 1 FOR UPDATE SKIP LOCKED) + 1 AS int), 0)
                         ELSE
                           -1
-                        END AS Id FROM {{fullTableName}};
+                        END;
                         """;
     }