Snails

Class: Snails::QueueJob

Inherits:
Sqlx::Base show all
Defined in:
runtime/snails/queue_job.rbs,
runtime/snails/queue_job.rb

Constant Summary collapse

COLUMNS =

The projected columns — every column the queue row carries, in DDL order, with a reader generated for each (the hash-backed accessor loop: spinel unrolls a literal array in the class body). The list is the whole projection contract — only_columns derives from it and the readers read through read_attribute — so a new DDL column joins it in one place; the harness contract test compares it with the template's PRAGMA table_info, and the declared type exceptions below must stay inside it.

%w[id name payload status scheduled_at started_at finished_at attempts last_error created_at updated_at].each { |column| define_method(column) { read_attribute(column) } }

Constants inherited from Sqlx::Base

Sqlx::Base::TIMESTAMP_COLUMNS

Instance Attribute Summary collapse

Class Method Summary collapse

Methods inherited from Sqlx::Base

all, boolean_columns, date_columns, from_row, transaction

Methods included from Sqlx::Query

#count, #create!, #delete!, #delete_all!, #exists?, #fetch_many, #find, #find_by, #find_by!, #find_by_sql, #find_each, #find_in_batches, #projection, #row_by_sql, #rows_by_sql, #update!, #update_all!, #upsert_all!, #where_clause, #write

Methods included from Sqlx::Model

#enum

Methods included from Sqlx::Row

#initialize, #read_attribute

Instance Attribute Details

#attempts ⇒ Integer (readonly)

Returns the value of attribute attempts.

Returns:



12
13
14
 
# File 'runtime/snails/queue_job.rbs', line 12

def attempts
  @attempts
end

#created_at ⇒ Time? (readonly)

Returns the value of attribute created_at.

Returns:



14
15
16
 
# File 'runtime/snails/queue_job.rbs', line 14

def created_at
  @created_at
end

#finished_at ⇒ Time? (readonly)

Returns the value of attribute finished_at.

Returns:



11
12
13
 
# File 'runtime/snails/queue_job.rbs', line 11

def finished_at
  @finished_at
end

#id ⇒ Integer (readonly)

Returns the value of attribute id.

Returns:



5
6
7
 
# File 'runtime/snails/queue_job.rbs', line 5

def id
  @id
end

#last_error ⇒ String? (readonly)

Returns the value of attribute last_error.

Returns:



13
14
15
 
# File 'runtime/snails/queue_job.rbs', line 13

def last_error
  @last_error
end

#name ⇒ String (readonly)

Returns the value of attribute name.

Returns:



6
7
8
 
# File 'runtime/snails/queue_job.rbs', line 6

def name
  @name
end

#payload ⇒ Object (readonly)

Returns the value of attribute payload.

Returns:



7
8
9
 
# File 'runtime/snails/queue_job.rbs', line 7

def payload
  @payload
end

#scheduled_at ⇒ Time? (readonly)

Returns the value of attribute scheduled_at.

Returns:



9
10
11
 
# File 'runtime/snails/queue_job.rbs', line 9

def scheduled_at
  @scheduled_at
end

#started_at ⇒ Time? (readonly)

Returns the value of attribute started_at.

Returns:



10
11
12
 
# File 'runtime/snails/queue_job.rbs', line 10

def started_at
  @started_at
end

#status ⇒ String (readonly)

Returns the value of attribute status.

Returns:



8
9
10
 
# File 'runtime/snails/queue_job.rbs', line 8

def status
  @status
end

#updated_at ⇒ Time? (readonly)

Returns the value of attribute updated_at.

Returns:



15
16
17
 
# File 'runtime/snails/queue_job.rbs', line 15

def updated_at
  @updated_at
end

Class Method Details

.claim_due(now) ⇒ Snails::QueueJob?

atomically claim the oldest pending row whose time has come: the pick and the state change are one UPDATE (with RETURNING), so two workers can never carry the same row, and a row scheduled for later stays untouched. The outer status = 'pending' guard is what makes the claim atomic under READ COMMITTED: a second claim that waited on the row lock re-evaluates the outer condition and sees running instead of claiming again. The returned instance carries the projected id, name, payload, and attempts — what a drain needs to run the job and name that run in its log.

Parameters:

  • now (Time) —

    the claim clock — rows scheduled after it stay

Returns:



68
 
# File 'runtime/snails/queue_job.rb', line 68

def self.claim_due: (Time now) -> Snails::QueueJob?

.datetime_columns ⇒ Object

Time columns: the storage time comes back as a Time (the built-in timestamp pair plus the queue's own three). The list replaces the default wholesale, so the pair is spelled again.



32
 
# File 'runtime/snails/queue_job.rb', line 32

def self.datetime_columns = %w[scheduled_at started_at finished_at created_at updated_at]

.enqueue!(name, payload, scheduled_at, now) ⇒ void

write one pending row. The payload is the logical value — the write entry encodes the declared JSON column before it reaches the database.

Parameters:

  • name (String) —

    the job's name — the stored constant path

  • payload (Object, nil) —

    the serializable arguments (nil for a payload-less job)

  • scheduled_at (Time) —

    when the row becomes due — now for a plain hand-off, the run time for a delayed one, the interval's start for a recurring occurrence

  • now (Time) —

    the created/updated stamp for the row



51
 
# File 'runtime/snails/queue_job.rb', line 51

def self.enqueue!: (String name, untyped? payload, Time scheduled_at, Time now) -> void

.json_columns ⇒ Object

payload stores JSON as text — the DDL column type is TEXT, and DDL alone cannot say "JSON" — so the codec pair is a declaration here: write entries encode at the entry, the default from_row decodes, and callers only ever see the logical value.



38
 
# File 'runtime/snails/queue_job.rb', line 38

def self.json_columns = %w[payload]

.mark_failed!(id, error, now) ⇒ void

record a failed run on its row, with the error message: an unresolvable name and a raising body both land here, so a failure never passes silently.

Parameters:

  • id (Integer) —

    the row's id

  • error (String) —

    the failure message

  • now (Time) —

    the finish stamp



112
 
# File 'runtime/snails/queue_job.rb', line 112

def self.mark_failed!: (untyped id, String error, Time now) -> void

.mark_succeeded!(id, now) ⇒ void

record a finished run on its row: the status and the finish stamp make the row the execution history.

Parameters:

  • id (Integer) —

    the row's id

  • now (Time) —

    the finish stamp



100
 
# File 'runtime/snails/queue_job.rb', line 100

def self.mark_succeeded!: (untyped id, Time now) -> void

.only_columns ⇒ Object 



27
 
# File 'runtime/snails/queue_job.rb', line 27

def self.only_columns = COLUMNS.map { |c| c.to_sym }

.prune_finished!(finished_before, limit:) ⇒ Integer

retention delete in bounded batches: finished rows (succeeded/failed) past the window go, pending rows and recent history stay. One call removes at most limit rows, so a large backlog never holds the write lock for its whole drain — the caller repeats until a short batch says the backlog is gone (Snails::QueuePruneJob owns the loop).

The statement is raw SQL because SQLite has no portable DELETE ... LIMIT (the bundled build omits the compile option), while IN (SELECT ... LIMIT) works on both backends. The inner SELECT stays unordered on purpose: with an ORDER BY, SQLite sorts every matching row before applying LIMIT, which would read the whole match set anyway.

Parameters:

  • finished_before (Time) —

    the retention cutoff

  • limit (Integer) —

    the most rows one call deletes (positive)

Returns:

  • (Integer) —

    the number of rows deleted (0 when none matched)

Raises:

  • (ArgumentError) —

    limit is not a positive Integer



133
 
# File 'runtime/snails/queue_job.rb', line 133

def self.prune_finished!: (Time finished_before, limit: Integer) -> Integer

.release_orphaned!(started_before, now) ⇒ Integer

return rows claimed by a dead process to pending: a claim older than the cutoff is the only evidence needed (at-least-once — the attempts counter keeps the lost run).

Parameters:

  • started_before (Time) —

    the claim-age cutoff

  • now (Time) —

    the updated stamp for the released rows

Returns:

  • (Integer) —

    the number of rows returned to pending



89
 
# File 'runtime/snails/queue_job.rb', line 89

def self.release_orphaned!: (Time started_before, Time now) -> Integer

.table_name ⇒ Object 



15
 
# File 'runtime/snails/queue_job.rb', line 15

def self.table_name = "snails_queue_jobs"