Merge pull request #231 from collectiveidea/fixup-skip-locked

Fixup reserve_with_scope_using_optimized_postgres

.github/workflows/ci.yml

@@ -216,6 +216,7 @@ jobs:
       with:
         ruby-version: ${{ matrix.ruby }}
         bundler-cache: true # runs 'bundle install' and caches installed gems automatically
+        cache-version: 2
     - name: Run tests
       env:
         BUNDLE_GEMFILE: ${{ matrix.gemfile }}

lib/delayed/backend/active_record.rb

@@ -133,19 +133,23 @@ def self.reserve_with_scope_using_optimized_postgres(ready_scope, worker, now)
           # Note: active_record would attempt to generate UPDATE...LIMIT like
           # SQL for Postgres if we use a .limit() filter, but it would not
           # use 'FOR UPDATE' and we would have many locking conflicts
+          subquery = ready_scope.limit(1).lock(true).select("id").to_sql
+
           # On PostgreSQL >= 9.5 we leverage SKIP LOCK to avoid multiple workers blocking each other
           # when attempting to get the next available job
           # https://www.postgresql.org/docs/9.5/sql-select.html#SQL-FOR-UPDATE-SHARE
-          skip_locked = ""
-          if connection.respond_to?(:postgresql_version) && connection.postgresql_version >= 90500
-            skip_locked = " SKIP LOCKED"
+          if connection.send(:postgresql_version) >= 9_05_00 # rubocop:disable Style/NumericLiterals
+            subquery += " SKIP LOCKED"
           end
 
           quoted_name = connection.quote_table_name(table_name)
-          subquery    = ready_scope.limit(1).lock(true).select("id").to_sql
-          sql         = "UPDATE #{quoted_name} SET locked_at = ?, locked_by = ? WHERE id IN (#{subquery}#{skip_locked}) RETURNING *"
-          reserved    = find_by_sql([sql, now, worker.name])
-          reserved[0]
+          find_by_sql(
+            [
+              "UPDATE #{quoted_name} SET locked_at = ?, locked_by = ? WHERE id IN (#{subquery}) RETURNING *",
+              now,
+              worker.name
+            ]
+          ).first
         end
 
         def self.reserve_with_scope_using_optimized_mysql(ready_scope, worker, now)