Class: Async::Background::Queue::Store

Inherits:
Object
  • Object
show all
Includes:
Clock
Defined in:
lib/async/background/queue/store.rb,
lib/async/background/queue/schema.rb

Defined Under Namespace

Classes: SchemaError

Constant Summary collapse

SCHEMA_VERSION = 
Schema::VERSION
MIGRATION_BUSY_TIMEOUT_MS = 
Schema::MIGRATION_BUSY_TIMEOUT_MS
REQUIRED_INDEXES = 
Schema::REQUIRED_INDEXES
SCHEMA = 
SQL::CREATE_SCHEMA
CLEANUP_INTERVAL = 
300
CLEANUP_AGE = 
3600
FAILED_RETENTION_AGE = 
7 * 24 * 3600
CLEANUP_VACUUM_THRESHOLD = 
100
ERROR_MESSAGE_MAX_LEN = 
2_000
EMPTY_ARGS_JSON = 
'[]'.freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path: self.class.default_path, options: {}) ⇒ Store

Returns a new instance of Store.



49
50
51
52
53
54
55
56
 
# File 'lib/async/background/queue/store.rb', line 49

def initialize(path: self.class.default_path, options: {})
  @path = path
  @options = StoreOptions.build(options)
  @pragma_sql = @options.pragma_sql.freeze
  @db = nil
  @schema_checked = false
  @last_cleanup_at = nil
end

Instance Attribute Details

#options ⇒ Object (readonly)

Returns the value of attribute options.



29
30
31
 
# File 'lib/async/background/queue/store.rb', line 29

def options
  @options
end

#path ⇒ Object (readonly)

Returns the value of attribute path.



29
30
31
 
# File 'lib/async/background/queue/store.rb', line 29

def path
  @path
end

Class Method Details

.default_path ⇒ Object 



47
 
# File 'lib/async/background/queue/store.rb', line 47

def self.default_path = 'async_background_queue.db'

.migrate!(path: default_path, options: {}) ⇒ Object 



31
32
33
34
35
36
37
 
# File 'lib/async/background/queue/store.rb', line 31

def self.migrate!(path: default_path, options: {})
  store = new(path: path, options: options)
  store.migrate!
  SCHEMA_VERSION
ensure
  store&.close
end

.prepare_dashboard!(path: default_path, options: {}) ⇒ Object 



39
40
41
42
43
44
45
 
# File 'lib/async/background/queue/store.rb', line 39

def self.prepare_dashboard!(path: default_path, options: {})
  store = new(path: path, options: options)
  store.prepare_dashboard!
  SCHEMA_VERSION
ensure
  store&.close
end

Instance Method Details

#close ⇒ Object 



177
178
179
180
181
182
183
184
185
 
# File 'lib/async/background/queue/store.rb', line 177

def close
  return unless connected?

  finalize_statements
  @db.execute(SQL::OPTIMIZE) rescue nil
  @db.close
  @db = nil
  @schema_checked = false
end

#complete(job_id, claim_token:, finished_at: realtime_now, duration_ms: nil) ⇒ Object 



123
124
125
126
127
128
129
130
131
132
 
# File 'lib/async/background/queue/store.rb', line 123

def complete(job_id, claim_token:, finished_at: realtime_now, duration_ms: nil)
  ensure_connection
  stepped(@complete_stmt) do |statement|
    statement.bind_param(1, finished_at)
    statement.bind_param(2, duration_ms)
    statement.bind_param(3, job_id)
    statement.bind_param(4, claim_token)
  end
  @db.changes.positive?
end

#data_version ⇒ Object 



172
173
174
175
 
# File 'lib/async/background/queue/store.rb', line 172

def data_version
  ensure_connection
  @db.get_first_value(SQL::DATA_VERSION).to_i
end

#enqueue(class_name, args = EMPTY_ARGS, run_at = nil, options: EMPTY_OPTIONS) ⇒ Object 



81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/async/background/queue/store.rb', line 81

def enqueue(class_name, args = EMPTY_ARGS, run_at = nil, options: EMPTY_OPTIONS)
  ensure_connection
  now = realtime_now
  stepped(@enqueue_stmt) do |statement|
    statement.bind_param(1, class_name)
    statement.bind_param(2, dump_args(args))
    statement.bind_param(3, dump_options(options))
    statement.bind_param(4, now)
    statement.bind_param(5, run_at || now)
  end
  @db.last_insert_row_id
end

#fail(job_id, claim_token:, error_class: nil, error_message: nil, finished_at: realtime_now, duration_ms: nil) ⇒ Object 



134
135
136
137
138
 
# File 'lib/async/background/queue/store.rb', line 134

def fail(job_id, claim_token:, error_class: nil, error_message: nil, finished_at: realtime_now, duration_ms: nil)
  ensure_connection
  bind_failure(@fail_stmt, finished_at, duration_ms, error_class, error_message, job_id, claim_token)
  @db.changes.positive?
end

#fetch(worker_id) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/async/background/queue/store.rb', line 94

def fetch(worker_id)
  ensure_connection
  token = generate_claim_token
  now = realtime_now

  row = transaction do
    stepped(@fetch_stmt) do |statement|
      statement.bind_param(1, worker_id)
      statement.bind_param(2, now)
      statement.bind_param(3, token)
      statement.bind_param(4, now)
    end
  end
  return if row.nil? || row.empty?

  maybe_cleanup
  job_from_row(row, token)
end

#mark_started!(job_id, claim_token:, started_at: realtime_now) ⇒ Object 



113
114
115
116
117
118
119
120
121
 
# File 'lib/async/background/queue/store.rb', line 113

def mark_started!(job_id, claim_token:, started_at: realtime_now)
  ensure_connection
  stepped(@mark_started_stmt) do |statement|
    statement.bind_param(1, started_at)
    statement.bind_param(2, job_id)
    statement.bind_param(3, claim_token)
  end
  @db.changes.positive?
end

#migrate! ⇒ Object Also known as: ensure_database!

Raises:



58
59
60
61
62
63
64
 
# File 'lib/async/background/queue/store.rb', line 58

def migrate!
  raise SchemaError, 'close the Store before calling migrate!' if connected?

  with_database { |db| migrate_database!(db) }
  @schema_checked = true
  self
end

#next_pending_run_at ⇒ Object 



167
168
169
170
 
# File 'lib/async/background/queue/store.rb', line 167

def next_pending_run_at
  ensure_connection
  stepped(@next_pending_stmt)&.first
end

#prepare_dashboard! ⇒ Object

Raises:



68
69
70
71
72
73
74
 
# File 'lib/async/background/queue/store.rb', line 68

def prepare_dashboard!
  raise SchemaError, 'close the Store before calling prepare_dashboard!' if connected?

  with_database { |db| Schema.prepare_dashboard!(db) }
  @schema_checked = true
  self
end

#recover(worker_id) ⇒ Object 



161
162
163
164
165
 
# File 'lib/async/background/queue/store.rb', line 161

def recover(worker_id)
  ensure_connection
  stepped(@requeue_stmt) { |statement| statement.bind_param(1, worker_id) }
  @db.changes
end

#retry_or_fail(job_id, claim_token:, error_class: nil, error_message: nil, fallback_options: nil, finished_at: realtime_now, duration_ms: nil) ⇒ Object 



140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/async/background/queue/store.rb', line 140

def retry_or_fail(
  job_id,
  claim_token:,
  error_class: nil,
  error_message: nil,
  fallback_options: nil,
  finished_at: realtime_now,
  duration_ms: nil
)
  ensure_connection

  transaction do
    stored_options = stored_options_for(job_id, claim_token)
    next unless lease_alive?(job_id, claim_token)

    policy = retry_policy(stored_options, fallback_options)
    policy_retries?(policy) ? retry_job!(job_id, claim_token, policy, error_class, error_message) :
      fail_job!(job_id, claim_token, error_class, error_message, finished_at, duration_ms)
  end
end

#schema_version ⇒ Object 



76
77
78
79
 
# File 'lib/async/background/queue/store.rb', line 76

def schema_version
  ensure_connection
  @db.get_first_value(SQL::USER_VERSION).to_i
end