Add a new adapter to work the job with multiple database #103

ankithads · 2024-06-13T16:17:20Z

What?
Added a new adapter which uses 2 databases to work the job.
database 1 will have the job enqueued and will fetch the jobs to be worked
database 2 will be used to acquire the advisory lock on the job.

Gemfile

ameykusurkar

Done a first pass of the implementation, looking sensible. Suggested some improvements

lib/que/adapters/yugabyte.rb

lib/que.rb

lib/que/adapters/yugabyte.rb

ameykusurkar · 2024-06-26T13:16:01Z

lib/que/adapters/yugabyte.rb

+
+             if locked?(result.first['job_id'])
+                return result
+             end


An idea I had. We can do something like this to make this a bit more efficient, basically we use the row itself as a temporary lock to prevent a lot of race conditions. This way, a lot of the work is done within the main database itself.

Let me know what you think.

loop do job_connection.transaction do # This locks the row only while we are checking the advisory lock database. # This saves duplicate attempts by different works to acquire the lock. result = Que.execute("SELECT ... FOR UPDATE SKIPPED LOCK") return result if result.empty? if locked?(result.first['job_id']) return result end # lock is released after the advisory lock check as the transaction completes end end

we will have cross database transaction in this case. If that's not an issue then this would be good

we will have cross database transaction in this case

Cross database transactions are mainly a problem when you're trying to maintain consistency of writes between two databases (eg. if one commits but the other rolls back).

I don't think this is a problem here, so should be safe to ignore: technically acquiring the a lock is a write, but the SELECT FOR UPDATE is temporary and not the source of truth, and the lock is released at the end of the transaction anyway.

lib/que/adapters/yugabyte.rb

lib/que.rb

lib/que/adapters/active_record_with_lock.rb

spec/spec_helper.rb

ameykusurkar

Looking great! Just some minor stylistic stuff and I think it's good to go

ameykusurkar · 2024-07-04T17:13:15Z

lib/que/adapters/active_record_with_lock.rb

+
+        def unlock_job(job_id)
+            lock_database_connection.execute("SELECT pg_advisory_unlock(#{job_id})")
+        end


The reason we don't need any error handling in this class is because the locker will release the lock on failure, is that correct?

yeah. Also I have also added additional comments on this on what happens when the connection goes bad and someother connection tries to release the lock. In this case it will just no-op and return false as the connection would be already released when the previous connection goes bad

ameykusurkar · 2024-07-04T17:33:27Z

lib/que/adapters/active_record_with_lock.rb

+module Que
+    module Adapters
+      class ActiveRecordWithLock < Que::Adapters::ActiveRecord
+        attr_accessor :job_connection_pool, :lock_record


Don't think we are using the attr_accessor anywhere, can remove the line in that case

ameykusurkar · 2024-07-04T17:34:29Z

lib/que/adapters/active_record_with_lock.rb

+                result = Que.execute(:find_job_to_lock, [queue, cursor])
+                break if result.empty?
+                cursor = result.first['job_id']
+                if pg_try_advisory_lock?(cursor)


Suggested change

if pg_try_advisory_lock?(cursor)

break if pg_try_advisory_lock?(cursor)

ameykusurkar · 2024-07-04T17:35:48Z

lib/que/locker.rb

@@ -60,7 +60,6 @@ def initialize(queue:, cursor_expiry:, window: nil, budget: nil, secondary_queue
      @queue_expires_at = {}
      @secondary_queues = secondary_queues
      @consolidated_queues = Array.wrap(queue).concat(secondary_queues)
-


I'm seeing a bunch of diffs that are just changing whitespace/formatting. Should we just get rubocop to standardise all of this?

lib/que/sql.rb

ameykusurkar · 2024-07-04T17:37:55Z

spec/lib/que/locker_spec.rb

@@ -117,7 +122,7 @@ def expect_to_lock_with(cursor:)
            expect_to_lock_with(cursor: job_1[:job_id])
            expect_to_work(job_2)

-            @epoch += cursor_expiry # our cursor should now expire
+            @epoch += (cursor_expiry) # our cursor should now expire


Why are the brackets are needed here?

will remove all these. I was trying out something for flaky tests.

spec/spec_helper.rb

ankithads force-pushed the add-lock-databse branch 8 times, most recently from d92a84a to 59aab33 Compare June 18, 2024 15:25

ankithads force-pushed the add-lock-databse branch 3 times, most recently from 5c6353d to f29a8b8 Compare June 25, 2024 09:52

ankithads commented Jun 25, 2024

View reviewed changes

Gemfile Outdated Show resolved Hide resolved

ankithads force-pushed the add-lock-databse branch 3 times, most recently from e1f6f21 to 31bf0fe Compare June 25, 2024 15:52

ameykusurkar reviewed Jun 26, 2024

View reviewed changes

ankithads force-pushed the add-lock-databse branch 5 times, most recently from d872b0e to 6b3bac9 Compare July 2, 2024 14:48

ameykusurkar reviewed Jul 2, 2024

View reviewed changes

ankithads added 8 commits July 4, 2024 08:48

add github workflows

0fd2b0f

fix flaky specs

c36e280

use env variable

58cdde6

add yugabyte adapter

7e108ad

use yugabyte adapter for getting the lock connection

2f281c6

pass the connection pool on initialise the adapter

89b3c7f

convert recursion to loop

df83d51

remove connection pool for lock connection

a9d95e3

fix flaky specs

1569cba

ankithads force-pushed the add-lock-databse branch from c5bd47a to 30ee504 Compare July 4, 2024 11:13

refactor

adbbcb6

ankithads force-pushed the add-lock-databse branch from 30ee504 to adbbcb6 Compare July 4, 2024 11:17

ankithads marked this pull request as ready for review July 4, 2024 11:18

ankithads changed the title ~~add lock databse~~ Add a new adapter to work the job with multiple database Jul 4, 2024

ameykusurkar reviewed Jul 4, 2024

View reviewed changes

ankithads added 2 commits July 5, 2024 09:24

add additional comments

c291312

fix formatting

4623fab

ankithads force-pushed the add-lock-databse branch from 9fb3896 to 4623fab Compare July 5, 2024 09:53

add a prefix to job_id while locking

cc8d50b

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add a new adapter to work the job with multiple database #103

Add a new adapter to work the job with multiple database #103

ankithads commented Jun 13, 2024 •

edited

Loading

ameykusurkar left a comment

ameykusurkar Jun 26, 2024

ankithads Jun 27, 2024

ameykusurkar Jun 28, 2024 •

edited

Loading

ameykusurkar left a comment

ameykusurkar Jul 4, 2024

ankithads Jul 5, 2024

ameykusurkar Jul 4, 2024

ameykusurkar Jul 4, 2024

ameykusurkar Jul 4, 2024

ameykusurkar Jul 4, 2024

ankithads Jul 5, 2024

	if pg_try_advisory_lock?(cursor)
	break if pg_try_advisory_lock?(cursor)

Add a new adapter to work the job with multiple database #103

Are you sure you want to change the base?

Add a new adapter to work the job with multiple database #103

Conversation

ankithads commented Jun 13, 2024 • edited Loading

ameykusurkar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ameykusurkar Jun 28, 2024 • edited Loading

Choose a reason for hiding this comment

ameykusurkar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ankithads commented Jun 13, 2024 •

edited

Loading

ameykusurkar Jun 28, 2024 •

edited

Loading