DEV: Fix job cluster concurrency spec timing out (#25035)

Why this change?

On CI, we have been seeing the "handles job concurrency" job timing out
on CI after 45 seconds. Upon closer inspection of `Jobs::Base#perform`
when cluster concurrency has been set, we see that a thread is spun up
to extend the expiring of a redis key by 120 seconds every 60 seconds
while the job is still being executed. The thread looks like this before
the fix:

```
keepalive_thread =
  Thread.new do
    while parent_thread.alive? && !finished
      Discourse.redis.without_namespace.expire(cluster_concurrency_redis_key, 120)
      sleep 60
    end
  end
```

In an ensure block of `Jobs::Base#perform`, the thread is stop by doing
something like this:

```
finished = true
keepalive_thread.wakeup
keepalive_thread.join
```

If the thread is sleeping, `keepalive_thread.wakeup` will stop the
`sleep` method and run the next iteration causing the thread to
complete. However, there is a timing issue at play here. If
`keepalive_thread.wakeup` is called at a time when the thread is not
sleeping, it will have no effect and the thread may end up sleeping for
60 seconds which is longer than our timeout on CI of 45 seconds.

What does this change do?

1. Change `sleep 60` to sleep in intervals of 1 second checking if the
   job has been finished each time.

2. Add `use_redis_snapshotting` to `Jobs::Base` spec since Redis is
   involved in scheduling and we want to ensure we don't leak Redis
keys.

3. Add `ConcurrentJob.stop!` and `thread.join` to `ensure` block in "handles job concurrency"
   test since a failing expectation will cause us to not clean up the
thread we created in the test.
This commit is contained in:
Alan Guo Xiang Tan 2023-12-26 14:47:03 +08:00 committed by GitHub
parent 89705be722
commit 043ba1d179
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 10 additions and 4 deletions

View File

@ -255,7 +255,12 @@ module Jobs
Thread.new do
while parent_thread.alive? && !finished
Discourse.redis.without_namespace.expire(cluster_concurrency_redis_key, 120)
sleep 60
# Sleep for 60 seconds, but wake up every second to check if the job has been completed
60.times do
break if finished
sleep 1
end
end
end
end

View File

@ -1,6 +1,8 @@
# frozen_string_literal: true
RSpec.describe ::Jobs::Base do
use_redis_snapshotting
class GoodJob < ::Jobs::Base
attr_accessor :count
def execute(args)
@ -62,12 +64,11 @@ RSpec.describe ::Jobs::Base do
wait_for { ConcurrentJob.running? }
ConcurrentJob.new.perform({ "test" => 100 })
expect(Sidekiq::Queues["default"].size).to eq(1)
expect(Sidekiq::Queues["default"][0]["args"][0]).to eq("test" => 100)
ensure
ConcurrentJob.stop!
thread.join
end