Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions app/jobs/salesforce/salesforce_sync_job.rb
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,14 @@ module Salesforce
class SalesforceSyncJob < ApplicationJob
include GoodJob::ActiveJobExtensions::Concurrency

self.good_job_labels = [-> { "#{self.class.name}/#{concurrency_key_id}" }]

# Serialise concurrent performs for the same record (same class + record ID)
# to prevent TOCTOU races on find_or_initialize_by + save!, while allowing
# jobs for different records to run fully in parallel.
good_job_control_concurrency_with(
good_job_concurrency_rule(
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
Resolved
perform_limit: 1,
key: -> { "#{self.class.name}/#{concurrency_key_id}" }
label: -> { "#{self.class.name}/#{concurrency_key_id}" }
)

class SalesforceRecordNotFound < StandardError
Expand Down
7 changes: 7 additions & 0 deletions db/migrate/20261106162610_add_lock_type_to_good_jobs.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
# frozen_string_literal: true

class AddLockTypeToGoodJobs < ActiveRecord::Migration[8.1]
def change
add_column :good_jobs, :lock_type, :integer, limit: 2 unless column_exists?(:good_jobs, :lock_type)
end
end
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
# frozen_string_literal: true

class AddIndexGoodJobsForCandidateDequeueUnlocked < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_jobs, [:priority, :scheduled_at, :id],
name: :index_good_jobs_for_candidate_dequeue_unlocked,
order: { priority: "ASC NULLS LAST", scheduled_at: :asc, id: :asc },
where: "finished_at IS NULL AND locked_by_id IS NULL",
algorithm: :concurrently,
if_not_exists: true
end
end
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
# frozen_string_literal: true

class AddIndexGoodJobsPriorityScheduledAt < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
reversible do |dir|
dir.up do
# Ensure this incremental update migration is idempotent
# with monolithic install migration.
return if connection.index_name_exists?(:good_jobs, "index_good_jobs_on_priority_scheduled_at_unfinished") &&
connection.index_name_exists?(:good_jobs, "index_good_jobs_on_queue_name_priority_scheduled_at_unfinished")
end
end

add_index :good_jobs, [:priority, :scheduled_at, :id],
where: "finished_at IS NULL", name: "index_good_jobs_on_priority_scheduled_at_unfinished",
algorithm: :concurrently
add_index :good_jobs, [:queue_name, :scheduled_at, :id],
where: "finished_at IS NULL", name: "index_good_jobs_on_queue_name_priority_scheduled_at_unfinished",
algorithm: :concurrently
end
end
12 changes: 12 additions & 0 deletions db/migrate/20261106162613_add_index_good_jobs_queue_name.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
# frozen_string_literal: true

class AddIndexGoodJobsQueueName < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_jobs, :queue_name,
name: :index_good_jobs_on_queue_name,
algorithm: :concurrently,
if_not_exists: true
end
end
12 changes: 12 additions & 0 deletions db/migrate/20261106162614_add_index_good_jobs_created_at.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
# frozen_string_literal: true

class AddIndexGoodJobsCreatedAt < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_jobs, :created_at,
name: :index_good_jobs_on_created_at,
algorithm: :concurrently,
if_not_exists: true
end
end
14 changes: 14 additions & 0 deletions db/migrate/20261106162615_add_index_good_jobs_discarded.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
# frozen_string_literal: true

class AddIndexGoodJobsDiscarded < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_jobs, :finished_at,
name: :index_good_jobs_on_discarded,
order: { finished_at: :desc },
where: "finished_at IS NOT NULL AND error IS NOT NULL",
algorithm: :concurrently,
if_not_exists: true
end
end
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
# frozen_string_literal: true

class AddIndexGoodJobsScheduledAtAndQueueName < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_jobs, [:scheduled_at, :queue_name],
name: :index_good_jobs_on_scheduled_at_and_queue_name,
algorithm: :concurrently,
if_not_exists: true
end
end
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
# frozen_string_literal: true

class AddIndexGoodJobsOnUnfinishedOrErrored < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_jobs, :id,
name: :index_good_jobs_on_unfinished_or_errored,
where: "finished_at IS NULL OR error IS NOT NULL",
algorithm: :concurrently,
if_not_exists: true
end
end
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
# frozen_string_literal: true

class AddIndexGoodJobsDiscardedJobClass < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_jobs, [:job_class, :finished_at],
name: :index_good_jobs_on_discarded_job_class,
where: "finished_at IS NOT NULL AND error IS NOT NULL",
algorithm: :concurrently,
if_not_exists: true
end
end
21 changes: 21 additions & 0 deletions db/migrate/20261106162619_add_index_good_jobs_dequeue_by_queue.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
# frozen_string_literal: true

class AddIndexGoodJobsDequeueByQueue < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_jobs, [:queue_name, :priority, :created_at],
name: :index_good_jobs_dequeue_by_queue,
order: { priority: "ASC NULLS LAST", created_at: :asc },
where: "finished_at IS NULL AND locked_by_id IS NULL",
algorithm: :concurrently,
if_not_exists: true

add_index :good_jobs, [:queue_name, :priority, :scheduled_at],
name: :index_good_jobs_dequeue_by_queue_scheduled_at,
order: { priority: "ASC NULLS LAST", scheduled_at: :asc },
where: "finished_at IS NULL AND locked_by_id IS NULL",
algorithm: :concurrently,
if_not_exists: true
end
end
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
# frozen_string_literal: true

class AddIndexGoodJobExecutionsOnScheduledAt < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_job_executions, :scheduled_at,
name: :index_good_job_executions_on_scheduled_at,
algorithm: :concurrently,
if_not_exists: true
end
end
28 changes: 28 additions & 0 deletions db/migrate/20261106162621_add_batch_query_indexes.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
# frozen_string_literal: true

class AddBatchQueryIndexes < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def change
add_index :good_job_batches, :finished_at,
name: :index_good_job_batches_on_finished_at,
where: "finished_at IS NOT NULL",
algorithm: :concurrently,
if_not_exists: true
add_index :good_job_batches, [:created_at, :id],
name: :index_good_job_batches_on_created_at_and_id,
order: { created_at: :desc, id: :desc },
algorithm: :concurrently,
if_not_exists: true
add_index :good_jobs, :batch_id,
name: :index_good_jobs_on_batch_id_unfinished,
where: "batch_id IS NOT NULL AND finished_at IS NULL",
algorithm: :concurrently,
if_not_exists: true
add_index :good_jobs, :batch_callback_id,
name: :index_good_jobs_on_batch_callback_id_unfinished,
where: "batch_callback_id IS NOT NULL AND finished_at IS NULL",
algorithm: :concurrently,
if_not_exists: true
end
end
55 changes: 55 additions & 0 deletions db/migrate/20261106162622_remove_redundant_good_jobs_indexes.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
# frozen_string_literal: true

class RemoveRedundantGoodJobsIndexes < ActiveRecord::Migration[8.1]
disable_ddl_transaction!

def up
remove_index(
:good_jobs,
name: :index_good_jobs_jobs_on_priority_created_at_when_unfinished,
algorithm: :concurrently,
if_exists: true
)
remove_index(
:good_jobs,
name: :index_good_jobs_on_priority_scheduled_at_unfinished_unlocked,
algorithm: :concurrently,
if_exists: true
)
remove_index(
:good_jobs,
name: :index_good_jobs_on_queue_name_and_scheduled_at,
algorithm: :concurrently,
if_exists: true
)
end

def down
add_index(
:good_jobs,
[:priority, :created_at],
name: :index_good_jobs_jobs_on_priority_created_at_when_unfinished,
order: { priority: "DESC NULLS LAST", created_at: :asc },
where: "finished_at IS NULL",
algorithm: :concurrently,
if_not_exists: true
)
add_index(
:good_jobs,
[:priority, :scheduled_at],
name: :index_good_jobs_on_priority_scheduled_at_unfinished_unlocked,
order: { priority: "ASC NULLS LAST", scheduled_at: :asc },
where: "finished_at IS NULL AND locked_by_id IS NULL",
algorithm: :concurrently,
if_not_exists: true
)
add_index(
:good_jobs,
[:queue_name, :scheduled_at],
name: :index_good_jobs_on_queue_name_and_scheduled_at,
where: "finished_at IS NULL",
algorithm: :concurrently,
if_not_exists: true
)
end
end
51 changes: 51 additions & 0 deletions db/migrate/20261106162623_reorder_good_jobs_cron_index.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
# frozen_string_literal: true

class ReorderGoodJobsCronIndex < ActiveRecord::Migration[8.1]
CURRENT_INDEX = :index_good_jobs_on_cron_key_and_cron_at_cond
REPLACEMENT_INDEX = :index_good_jobs_on_cron_key_and_cron_at_desc_cond

disable_ddl_transaction!

def up
return if cron_index_has_expected_order?

add_index :good_jobs, [:cron_key, :cron_at],
name: REPLACEMENT_INDEX,
order: { cron_at: "DESC NULLS LAST" },
where: "cron_key IS NOT NULL",
unique: true,
algorithm: :concurrently,
if_not_exists: true
remove_index :good_jobs, name: CURRENT_INDEX, algorithm: :concurrently, if_exists: true
rename_index :good_jobs, REPLACEMENT_INDEX, CURRENT_INDEX
end

def down
return if cron_index_definition && !cron_index_has_expected_order?

add_index :good_jobs, [:cron_key, :cron_at],
name: REPLACEMENT_INDEX,
where: "cron_key IS NOT NULL",
unique: true,
algorithm: :concurrently,
if_not_exists: true
remove_index :good_jobs, name: CURRENT_INDEX, algorithm: :concurrently, if_exists: true
rename_index :good_jobs, REPLACEMENT_INDEX, CURRENT_INDEX
end

private

def cron_index_has_expected_order?
cron_index_definition&.include?("cron_at DESC NULLS LAST")
end

def cron_index_definition
connection.select_value(<<~SQL.squish)
SELECT indexdef
FROM pg_indexes
WHERE schemaname = ANY (current_schemas(false))
AND tablename = #{connection.quote('good_jobs')}
AND indexname = #{connection.quote(CURRENT_INDEX.to_s)}
SQL
end
end
27 changes: 27 additions & 0 deletions db/migrate/20261106162624_create_good_job_concurrency_claims.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
# frozen_string_literal: true

class CreateGoodJobConcurrencyClaims < ActiveRecord::Migration[8.1]
def up
# Ensure this incremental update migration is idempotent
# with monolithic install migration.
return if connection.table_exists?(:good_job_concurrency_claims)

create_table :good_job_concurrency_claims, id: :uuid do |t|
t.datetime :created_at, null: false
t.datetime :updated_at, null: false
t.uuid :job_id, null: false
t.uuid :locked_by_id
t.integer :state, limit: 2, null: false
t.text :key, null: false

t.index [:key, :job_id], unique: true
t.index [:key, :state, :created_at], name: :index_good_job_concurrency_claims_for_promotion
t.index :job_id
t.index :locked_by_id, where: "locked_by_id IS NOT NULL"
end
end

def down
drop_table :good_job_concurrency_claims, if_exists: true
end
end
Loading
Loading