Class: AceMQ::AMQP::Patterns::SQLIdempotencyStore

Inherits:
Object
  • Object
show all
Includes:
AceMQ::AMQP::Patterns::SQL::Statements
Defined in:
lib/acemq/amqp/patterns/sql_idempotency.rb

Overview

An idempotency store in a database table, shared by every consumer that points at it.

store = Patterns::SQLIdempotencyStore.new(connection: db)
store.create_schema                        # development and tests only

mq.consume("orders.new", &Patterns.idempotent(store) do |message|
payments.charge(message.payload)
Ack.accept
end)

The one InMemoryIdempotencyStore cannot replace. An in-process store deduplicates within one process, which is enough when duplicates arrive seconds apart on the same worker and useless the moment there are three workers behind one queue: the redelivery lands on a different one, finds an empty hash, and does the work again. Charging a card twice is not a caching problem.

The failure this has and the in-memory one does not

A shared store outlives the process using it, and that cuts both ways. If a consumer takes a key and then dies mid-handler, the row stays in the table. Naively, every redelivery of that message — for ever — is discarded as a duplicate, and a crash that should have cost one retry has silently deleted a message instead. The in-memory store never had this problem only because a crash wiped it.

So a key here is held under a lease, not a lock. It expires after claim_timeout, after which another consumer may take it over. That makes the timeout a real decision: too short and two consumers work on the same message at once while the first is still going; too long and a crashed consumer stalls that message for the duration. It should comfortably exceed the slowest handler, and the default is deliberately generous.

Confirmations are kept for retention and then forgotten, because a table remembering every identifier ever seen is a disk-space incident waiting to happen. A duplicate arriving after retention is handled again — retention is the window within which duplicates are actually expected, not a permanent record.

What it does not do

This deduplicates the delivery, not the work. If the handler writes to a different database from this table, a crash between the write committing and the confirmation landing leaves the work done and unrecorded, and the redelivery repeats it. Putting this table in the same database as the handler's writes closes that gap, and doing so is the caller's decision, since only the caller knows what the handler touches.

Schedule #purge_expired — hourly is ample. Nothing on the message path deletes anything, because a store that tidies up on the hot path makes every message pay for it.

Constant Summary collapse

DEFAULT_TABLE = 
"acemq_idempotency"
DEFAULT_CLAIM_TIMEOUT =

Long enough for a slow handler, short enough that a crash is not a permanent stall.

300.0
DEFAULT_RETENTION =

How long a confirmed identifier is remembered.

24 * 3600.0
CLAIMED =

Taken and being worked on.

"CLAIMED"
CONFIRMED =

Done, and remembered until retention runs out.

"CONFIRMED"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection:, retention: DEFAULT_RETENTION, claim_timeout: DEFAULT_CLAIM_TIMEOUT, table: DEFAULT_TABLE) ⇒ SQLIdempotencyStore

Returns a new instance of SQLIdempotencyStore.

Parameters:

  • connection (Object, #call) —

    where the table lives

  • retention (Numeric) (defaults to: DEFAULT_RETENTION) —

    seconds a confirmed key is remembered

  • claim_timeout (Numeric) (defaults to: DEFAULT_CLAIM_TIMEOUT) —

    seconds one consumer may hold a message before another may take it over. Must exceed the slowest handler, or two consumers will work on the same message at once.

  • table (String) (defaults to: DEFAULT_TABLE) —

    a plain SQL identifier

Raises:

  • (ArgumentError) 


125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 125

def initialize(connection:, retention: DEFAULT_RETENTION,
               claim_timeout: DEFAULT_CLAIM_TIMEOUT, table: DEFAULT_TABLE)
  @source = connection
  @table = SQL.table_name!(table)
  raise ArgumentError, "retention must be positive" unless retention.to_f.positive?
  unless claim_timeout.to_f.positive?
    raise ArgumentError, "claim_timeout must be positive"
  end

  @retention = retention.to_f
  @claim_timeout = claim_timeout.to_f
  # Identifies this instance's holds, so {#forget} can only give up its
  # own. Without it, releasing a failed handler's key would also delete
  # a key another worker had legitimately taken over after the lease
  # expired, and both would then run.
  @worker = SecureRandom.uuid
end

Instance Attribute Details

#table ⇒ Object (readonly)

The table this store reads and writes.



144
145
146
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 144

def table
  @table
end

#worker ⇒ Object (readonly)

Which worker this store claims as. Two stores on one table have different ones, which is what makes a release specific.



148
149
150
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 148

def worker
  @worker
end

Class Method Details

.ddl(table) ⇒ Object

Written here rather than in a .sql file, because the gem ships lib/**/*.rb and a schema missing from the package is a schema that works in the repository and nowhere after it.



99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 99

def self.ddl(table)
  [
    <<~SQL.chomp,
      CREATE TABLE IF NOT EXISTS #{table} (
        message_id  VARCHAR(255)             NOT NULL PRIMARY KEY,
        state       VARCHAR(16)              NOT NULL,
        claimed_by  VARCHAR(64),
        recorded_at TIMESTAMP WITH TIME ZONE NOT NULL,
        expires_at  TIMESTAMP WITH TIME ZONE NOT NULL
      )
    SQL
    # One column carries both deadlines because they are never both
    # meaningful: while a row is CLAIMED it holds the lease expiry, and
    # once CONFIRMED it holds the retention expiry. Both answer the same
    # question — is this row still binding? — so one index serves the
    # claim path and the purge.
    "CREATE INDEX IF NOT EXISTS #{table}_expiry ON #{table} (expires_at)"
  ]
end

Instance Method Details

#claim(key) ⇒ Symbol

Claims a message, or says why not.

Claim::CLAIMED when there was no row or its lease had run out; Claim::DUPLICATE when the row is confirmed; Claim::IN_PROGRESS when somebody holds a live, unconfirmed claim -- still working, or failed without releasing it. The difference between the last two is the difference between done and maybe-never-done, which is why AceMQ::AMQP::Patterns.idempotent accepts one and puts the other back.

Parameters:

  • key (String)

Returns:

  • (Symbol) —

    one of the Claim values



181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 181

def claim(key)
  # Every key that reaches a statement goes through SQL.key, and it is
  # not tidiness: a message id off the wire is ASCII-8BIT, sqlite3 binds
  # that as a blob, and SQLite never matches a blob against a text
  # value. Two strings that are == in Ruby were two rows in a table
  # whose primary key exists to make that impossible — so the same
  # message could be handled twice, which is the one failure this class
  # exists to prevent.
  key = SQL.key(key)
  now = Time.now.utc
  return Claim::CLAIMED if take(key, now)

  # Somebody already has a row. Taking it over is allowed only if their
  # hold has run out, and the guard lives in the WHERE clause so that
  # the check and the take-over are one statement — a select followed by
  # an update would let two consumers both pass the check and both
  # conclude they had won.
  return Claim::CLAIMED if taken_over?(key, now)

  # Only asked once the take-over lost, so it decides nothing about who
  # runs. No row means it was released or purged between the two
  # statements: in progress is the answer that cannot lose the message.
  state = run("SELECT state FROM #{@table} WHERE message_id = ?", [key]).rows.dig(0, 0)
  state == CONFIRMED ? Claim::DUPLICATE : Claim::IN_PROGRESS
end

#confirm(key) ⇒ Object

Records that the work is done, and starts the retention clock.

Called by AceMQ::AMQP::Patterns.idempotent after a handler accepts. Without it a lease would simply expire and the next duplicate would be handled again, which is the whole failure a shared store is for.



212
213
214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 212

def confirm(key)
  key = SQL.key(key)
  now = Time.now.utc
  updated = run(
    "UPDATE #{@table} SET state = ?, claimed_by = NULL, recorded_at = ?, " \
    "expires_at = ? WHERE message_id = ?",
    [CONFIRMED, SQL.at_utc(now), SQL.at_utc(now + @retention), key]
  ).affected
  # The row was purged, or the lease expired and somebody else took it.
  # The work still happened, so it is recorded either way: an unrecorded
  # confirmation is how the same charge gets made twice.
  insert_confirmed(key, now) if updated.to_i.zero?
  nil
end

#confirmed?(key) ⇒ Boolean

Whether a key is recorded as done and still within retention.

Returns:

  • (Boolean) 


240
241
242
243
244
245
246
247
248
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 240

def confirmed?(key)
  rows = run("SELECT expires_at FROM #{@table} WHERE message_id = ? AND state = ?",
             [SQL.key(key), CONFIRMED]).rows
  return false if rows.empty?

  # An expired row is answered as false rather than deleted: a read that
  # writes turns every duplicate check into a write on a shared table.
  SQL.time(rows[0][0]) > Time.now
end

#create_schema ⇒ Object

Creates the table if it is not already there.

For development and tests. In production the table belongs in the migration tool that owns the rest of the schema.



154
155
156
157
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 154

def create_schema
  self.class.ddl(@table).each { |statement| connection.run(statement, []) }
  nil
end

#first_time?(key) ⇒ Boolean

Records a key, and says whether this caller is the one to handle it.

The insert is the claim. A primary key makes it atomic across every process using this table, which is the entire reason a shared store can be trusted: two consumers racing on the same identifier cannot both succeed, whatever else they are doing.

Parameters:

  • key (String)

Returns:

  • (Boolean) —

    true for the caller that may do the work



168
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 168

def first_time?(key) = claim(key) == Claim::CLAIMED

#forget(key) ⇒ Object

Gives up this store's own hold, so a message that failed can be redone.

Only a live hold of this worker's, and never a confirmation. Deleting somebody else's would put two consumers on one message; deleting a confirmation would undo it.



233
234
235
236
237
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 233

def forget(key)
  run("DELETE FROM #{@table} WHERE message_id = ? AND state = ? AND claimed_by = ?",
      [SQL.key(key), CLAIMED, @worker])
  nil
end

#purge_expired ⇒ Integer

Deletes rows nobody is bound by any more: confirmations past their retention, and holds whose lease has run out.

Returns:

  • (Integer) —

    how many rows were removed



254
255
256
257
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 254

def purge_expired
  run("DELETE FROM #{@table} WHERE expires_at <= ?", [SQL.at_utc(Time.now)])
    .affected.to_i
end

#size ⇒ Object

How many identifiers the table holds, expired ones included.



260
261
262
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 260

def size
  run("SELECT COUNT(*) FROM #{@table}").rows.dig(0, 0).to_i
end

#to_s ⇒ Object 



264
 
# File 'lib/acemq/amqp/patterns/sql_idempotency.rb', line 264

def to_s = "SQLIdempotencyStore(#{@table})"