Module: DBOperations::Operations

Defined in:
app/workers/has_helpers/db_operations/operations.rb

Constant Summary collapse

MAX_PROCESSING_TIME = 
5.minutes
PAUSED_KEY = 
"db_ops_paused"
PAUSE_TABLE_KEY_PREFIX = 
"db_ops_pause_table"
DEFAULT_PAUSE_TABLE_DURATION_SECONDS = 
1.day.to_i

Class Method Summary collapse

Class Method Details

.connection_pool ⇒ Object

Use the master connection, otherwise this will break when running with crunchy-proxy. crunchy-proxy can not recongize that a function has writes in it unless you add it to an exception list.



93
94
95
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 93

def self.connection_pool
  HasHelpers::MasterRecord.connection_pool
end

.defer_processing(id) ⇒ Object 



55
56
57
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 55

def self.defer_processing(id)
  HasHelpers::PendingDBOperation.defer_processing(id)
end

.disabled_tables ⇒ Object 



130
131
132
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 130

def self.disabled_tables
  RedisConf.current.keys("#{PAUSE_TABLE_KEY_PREFIX}::*").map { |k| k.sub("#{PAUSE_TABLE_KEY_PREFIX}::", "") }
end

.enqueue_ops(except_for: nil) ⇒ Object 



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 18

def self.enqueue_ops(except_for: nil)
  pending_db_ops = HasHelpers::PendingDBOperation.where(enqueued_at: nil).order(:created_at).limit(1_000).select(:id)
  if except_for.present?
    pending_db_ops = pending_db_ops.where.not(relation_name: except_for)
  end

  pending_db_ops_sql = pending_db_ops.to_sql
  Enumerator.new do |yielder|
    holdstart = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)

    loop do
      enqueued_db_ops = HasHelpers::PendingDBOperation.uncached do
        HasHelpers::PendingDBOperation.find_by_sql <<-SQL
          WITH updated_rows AS (
            UPDATE pending_db_operations
            SET enqueued_at = now() at time zone 'utc'
            WHERE id IN (#{pending_db_ops_sql})
            RETURNING id
          )
          SELECT *
          FROM pending_db_operations
          INNER JOIN updated_rows ON updated_rows.id = pending_db_operations.id
        SQL
      end

      if enqueued_db_ops.present?
        yielder << enqueued_db_ops
      else
        break
      end

      endt = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
      break if (endt - holdstart) > MAX_PROCESSING_TIME
    end
  end
end

.pause(duration = 24.hours.to_i) ⇒ Object

Pause / Unpause #####



77
78
79
80
81
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 77

def self.pause(duration = 24.hours.to_i)
  ::HasUtils::Env.run_on_environment excluded: [::HasUtils::Env::TEST] do
    RedisConf.current.setex(PAUSED_KEY, duration, true)
  end
end

.pause_table(table_name, duration = DEFAULT_PAUSE_TABLE_DURATION_SECONDS) ⇒ Object

Use to indicate that we want to stop processing this table for the duration indicated. If no duration is specified, defaults to DEFAULT_PAUSE_TABLE_DURATION_SECONDS. Note that calling this overwrites any previous call to stop processing this table, even if the previous call's duration has not expired.



104
105
106
107
108
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 104

def self.pause_table(table_name, duration = DEFAULT_PAUSE_TABLE_DURATION_SECONDS)
  ::HasUtils::Env.run_on_environment excluded: [::HasUtils::Env::TEST] do
    RedisConf.current.setex(pause_table_redis_key(table_name), duration, true)
  end
end

.pause_table_redis_key(table_name) ⇒ Object 



126
127
128
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 126

def self.pause_table_redis_key(table_name)
  "#{PAUSE_TABLE_KEY_PREFIX}::#{table_name}"
end

.paused? ⇒ Boolean

Returns:

  • (Boolean) 


87
88
89
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 87

def self.paused?
  RedisConf.current.get(PAUSED_KEY).present?
end

.resume ⇒ Object 



83
84
85
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 83

def self.resume
  RedisConf.current.del(PAUSED_KEY)
end

.resume_all_tables ⇒ Object 



121
122
123
124
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 121

def self.resume_all_tables
  keys = RedisConf.current.keys("#{PAUSE_TABLE_KEY_PREFIX}::*")
  RedisConf.current.del(keys) if keys.present?
end

.resume_table(table_name) ⇒ Object

Cancels any pausing previously set for this table.



117
118
119
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 117

def self.resume_table(table_name)
  RedisConf.current.del(pause_table_redis_key(table_name))
end

.run_one_db_operation(id) ⇒ Object 



12
13
14
15
16
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 12

def self.run_one_db_operation(id)
  ::HasHelpers::DBOperations::Operations.with_advisory_lock_2("'pending_db_operations'::regclass::integer", id) do |connection|
    connection.execute("SELECT run_one_db_operation(#{id})")
  end
end

.table_paused?(table_name) ⇒ Boolean

Returns a boolean indicating if we are pausing processing of this table.

Returns:

  • (Boolean) 


111
112
113
114
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 111

def self.table_paused?(table_name)
  redis_key = pause_table_redis_key(table_name)
  RedisConf.current.get(redis_key).present?
end

.with_advisory_lock(key, connection_pool: self.connection_pool, &block) ⇒ Object

overrides to provide default connection pool



60
61
62
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 60

def self.with_advisory_lock(key, connection_pool: self.connection_pool, &block)
  ::PsqlHq::Utils.with_advisory_lock(key, connection_pool: connection_pool, &block)
end

.with_advisory_lock_2(key1, key2, connection_pool: self.connection_pool, &block) ⇒ Object

rubocop:disable Naming/VariableNumber



64
65
66
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 64

def self.with_advisory_lock_2(key1, key2, connection_pool: self.connection_pool, &block) # rubocop:disable Naming/VariableNumber
  ::PsqlHq::Utils.with_advisory_lock_2(key1, key2, connection_pool: connection_pool, &block)
end

.with_advisory_xact_lock(key, connection_pool: self.connection_pool, &block) ⇒ Object 



68
69
70
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 68

def self.with_advisory_xact_lock(key, connection_pool: self.connection_pool, &block)
  ::PsqlHq::Utils.with_advisory_xact_lock(key, connection_pool: connection_pool, &block)
end

.with_advisory_xact_lock_2(key1, key2, connection_pool: self.connection_pool, &block) ⇒ Object

rubocop:disable Naming/VariableNumber



72
73
74
 
# File 'app/workers/has_helpers/db_operations/operations.rb', line 72

def self.with_advisory_xact_lock_2(key1, key2, connection_pool: self.connection_pool, &block) # rubocop:disable Naming/VariableNumber
  ::PsqlHq::Utils.with_advisory_xact_lock_2(key1, key2, connection_pool: connection_pool, &block)
end