Class: Honker::Database

Inherits:
Object
  • Object
show all
Defined in:
lib/honker.rb

Overview

Database is a Honker handle over a SQLite file with the Honker extension loaded. The constructor bootstraps the schema; safe to open the same path from multiple processes.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path, extension_path: nil, watcher_backend: nil, watcher_poll_interval_ms: nil, extension_resolver: ExtensionResolver.new) ⇒ Database

Returns a new instance of Database.



179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
 
# File 'lib/honker.rb', line 179

def initialize(path, extension_path: nil, watcher_backend: nil,
               watcher_poll_interval_ms: nil,
               extension_resolver: ExtensionResolver.new)
  unless watcher_backend.nil? || watcher_backend.is_a?(String)
    raise ArgumentError, "unknown watcher backend"
  end
  unless watcher_poll_interval_ms.nil? || watcher_poll_interval_ms.to_i.positive?
    raise ArgumentError, "watcher_poll_interval_ms must be positive"
  end

  resolved_extension = extension_resolver.resolve(extension_path)
  @db = SQLite3::Database.new(path)
  @local_update_seq = 0
  @db.busy_timeout = 5000
  @db.execute("PRAGMA mmap_size = 0")
  @db.enable_load_extension(true)
  @db.load_extension(resolved_extension)
  @db.enable_load_extension(false)
  @db.execute_batch(DEFAULT_PRAGMAS)
  @db.execute("SELECT honker_bootstrap()")
  @watcher = CoreWatcher.new(path, resolved_extension, watcher_backend, watcher_poll_interval_ms)
end

Instance Attribute Details

#db ⇒ Object (readonly)

Returns the value of attribute db.



177
178
179
 
# File 'lib/honker.rb', line 177

def db
  @db
end

Instance Method Details

#close ⇒ Object 



202
203
204
205
 
# File 'lib/honker.rb', line 202

def close
  @watcher&.close
  @db&.close
end

#get_result(job_id) ⇒ Object

Fetch a stored result, or nil if absent or expired.



334
335
336
337
338
339
 
# File 'lib/honker.rb', line 334

def get_result(job_id)
  @db.get_first_row(
    "SELECT honker_result_get(?)",
    [job_id],
  )[0]
end

#mark_updated ⇒ Object 



207
208
209
 
# File 'lib/honker.rb', line 207

def mark_updated
  @local_update_seq += 1
end

#notify(channel, payload) ⇒ Object

Fire a pg_notify-style pub/sub signal. Returns the notification id.



256
257
258
259
 
# File 'lib/honker.rb', line 256

def notify(channel, payload)
  row = @db.get_first_row("SELECT notify(?, ?)", [channel, JSON.dump(payload)])
  row[0]
end

#notify_tx(tx, channel, payload) ⇒ Object

Fire a notification inside an open transaction. The signal lands only when the transaction commits.



263
264
265
266
267
268
269
 
# File 'lib/honker.rb', line 263

def notify_tx(tx, channel, payload)
  row = tx.query_row(
    "SELECT notify(?, ?)",
    [channel, JSON.dump(payload)],
  )
  row[0]
end

#outbox(name, delivery, visibility_timeout_s: 60, max_attempts: 5, base_backoff_s: 5) ⇒ Object

Transactional side-effect delivery built on a reserved queue.



233
234
235
236
237
238
239
240
241
242
 
# File 'lib/honker.rb', line 233

def outbox(name, delivery, visibility_timeout_s: 60, max_attempts: 5, base_backoff_s: 5)
  Outbox.new(
    self,
    name,
    delivery,
    visibility_timeout_s: visibility_timeout_s,
    max_attempts: max_attempts,
    base_backoff_s: base_backoff_s,
  )
end

#prune_notifications(older_than_s:) ⇒ Object

Delete notifications older than older_than_s seconds. Returns the number of rows deleted.



348
349
350
351
352
353
354
 
# File 'lib/honker.rb', line 348

def prune_notifications(older_than_s:)
  @db.execute(
    "DELETE FROM _honker_notifications WHERE created_at < unixepoch() - ?",
    [older_than_s],
  )
  @db.changes
end

#queue(name, visibility_timeout_s: 300, max_attempts: 3) ⇒ Object

Returns a Queue handle for a named queue.

visibility_timeout_s: 300   # claim expiry before reclaim
max_attempts:         3     # retries before moving to dead


223
224
225
226
227
228
229
230
 
# File 'lib/honker.rb', line 223

def queue(name, visibility_timeout_s: 300, max_attempts: 3)
  Queue.new(
    self,
    name,
    visibility_timeout_s: visibility_timeout_s,
    max_attempts: max_attempts,
  )
end

#save_result(job_id, value, ttl_s:) ⇒ Object

Persist a job result for later retrieval via get_result. value is stored verbatim — JSON-encode it yourself if you want to round-trip structured data.



325
326
327
328
329
330
331
 
# File 'lib/honker.rb', line 325

def save_result(job_id, value, ttl_s:)
  @db.get_first_row(
    "SELECT honker_result_save(?, ?, ?)",
    [job_id, value, ttl_s],
  )
  nil
end

#scheduler ⇒ Object

Returns the time-trigger Scheduler facade. Cheap — no allocation beyond the wrapper object.



251
252
253
 
# File 'lib/honker.rb', line 251

def scheduler
  Scheduler.new(self)
end

#stream(name) ⇒ Object

Returns a Stream handle for an append-only ordered log.



245
246
247
 
# File 'lib/honker.rb', line 245

def stream(name)
  Stream.new(self, name)
end

#sweep_rate_limits(older_than_s:) ⇒ Object

Sweep old rate-limit window rows. Returns count deleted.



315
316
317
318
319
320
 
# File 'lib/honker.rb', line 315

def sweep_rate_limits(older_than_s:)
  @db.get_first_row(
    "SELECT honker_rate_limit_sweep(?)",
    [older_than_s],
  )[0]
end

#sweep_results ⇒ Object

Drop expired result rows. Returns count swept.



342
343
344
 
# File 'lib/honker.rb', line 342

def sweep_results
  @db.get_first_row("SELECT honker_result_sweep()")[0]
end

#transaction ⇒ Object

Run a block inside a SQLite transaction. The block receives a Honker::Transaction; returning normally commits, raising rolls back, and tx.rollback! rolls back without surfacing an error.

db.transaction do |tx|
tx.execute("INSERT INTO orders ...")
q.enqueue_tx(tx, {order_id: 1})
end


279
280
281
282
283
284
285
286
287
288
289
290
291
 
# File 'lib/honker.rb', line 279

def transaction
  tx = Transaction.new(@db)
  begin
    @db.transaction do
      yield tx
    end
  rescue Transaction::Rollback
    # Caller used tx.rollback! to abort. The block exited with an
    # exception so the sqlite3 gem already rolled back — just
    # swallow the sentinel.
    nil
  end
end

#try_lock(name, owner:, ttl_s:) ⇒ Object

Try to acquire an advisory lock. Returns a Lock handle on success, nil if another owner holds it.



295
296
297
298
299
300
301
302
303
 
# File 'lib/honker.rb', line 295

def try_lock(name, owner:, ttl_s:)
  acquired = @db.get_first_row(
    "SELECT honker_lock_acquire(?, ?, ?)",
    [name, owner, ttl_s],
  )[0]
  return nil unless acquired == 1

  Lock.new(self, name, owner)
end

#try_rate_limit(name, limit:, per:) ⇒ Object

Fixed-window rate limiter. Returns true if this call fits within limit requests per per seconds.



307
308
309
310
311
312
 
# File 'lib/honker.rb', line 307

def try_rate_limit(name, limit:, per:)
  @db.get_first_row(
    "SELECT honker_rate_limit_try(?, ?, ?)",
    [name, limit, per],
  )[0] == 1
end

#update_snapshot ⇒ Object 



211
212
213
 
# File 'lib/honker.rb', line 211

def update_snapshot
  @local_update_seq
end

#wait_for_update(timeout_s) ⇒ Object 



215
216
217
 
# File 'lib/honker.rb', line 215

def wait_for_update(timeout_s)
  @watcher.wait(timeout_s)
end