Class: AceMQ::AMQP::Patterns::SQLOutboxStore

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

Overview

An outbox in a database table, in the same database as the work it accompanies.

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

db.transaction do
orders.insert(order)
store.add(Patterns.record(mq, event, to: "order.placed"), connection: db)
end

The connection handed to add is the point of the whole class. InMemoryOutboxStore is criticised in its own comment for not sharing a transaction with anybody's database, and a store that opened its own connection would have exactly that flaw while looking as though it had been dealt with: the insert would commit on its own, and a business write that rolled back afterwards would leave a message queued for something that never happened.

So the insert goes on the caller's connection, inside the caller's transaction, and this class neither commits it nor closes it. The message becomes durable exactly when the work does, and not before.

Two sources of connections

The two halves of this class run in different worlds. add belongs to the caller's transaction. Everything the relay does — claiming, marking, counting — happens on a background thread with no ambient transaction, so it wants a connection of its own. Under a connection pool relay: has to be a different connection from the one a request is using, or the relay's thread and the request's thread are writing down the same socket. With one connection — SQLite, a single-threaded service — the same one for both is correct and is the default.

Claiming is a lease, not a lock

A held lock lasts as long as its transaction, so a relay that dies mid-batch either strands its rows until the database notices the connection has gone or holds a transaction open across a network publish. Neither is acceptable. A lease is a timestamp: it expires on its own, however the holder died, and the rows return to circulation with nobody intervening.

The claim is taken in two statements, and the second is what makes it safe. Candidates are selected, then each is updated with the lease still conditional on being unclaimed. Two relays that select the same candidates both try to update them; row locking serialises the updates, the loser re-evaluates its condition against the committed row and matches nothing. The claim is decided by the update's row count, never by the select.

Constant Summary collapse

DEFAULT_TABLE = 
"acemq_outbox"
DEFAULT_MAX_ATTEMPTS =

How many failures a record may accumulate before it stops being claimed and stays for somebody to look at.

10
DEFAULT_LEASE =

How long a relay holds a record before another may take it over. Comfortably longer than a publish and a confirm, and short enough that a relay that died does not strand a batch for long.

60.0
MAX_BATCH =

More than this in one sweep is a relay that has stopped being a relay.

10_000
COLUMNS =

Positional, because two drivers disagree about hash keys and neither disagrees about order.

"id, exchange_name, routing_key, body, body_encoding, content_type, " \
"headers, created_at, attempts, last_error"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection:, relay: nil, table: DEFAULT_TABLE, max_attempts: DEFAULT_MAX_ATTEMPTS, lease: DEFAULT_LEASE) ⇒ SQLOutboxStore

Returns a new instance of SQLOutboxStore.

Parameters:

  • connection (Object, #call) —

    the caller's connection, or something that returns it — a lambda reading a thread local is the usual shape under a framework that binds one per request. What it must never be is something that opens a fresh connection: that is the flaw this class exists to avoid.

  • relay (Object, #call, nil) (defaults to: nil) —

    connections for the relay's own work; the same as connection when there is only one

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

    a plain SQL identifier

  • max_attempts (Integer) (defaults to: DEFAULT_MAX_ATTEMPTS) —

    failures before a record is left alone

  • lease (Numeric) (defaults to: DEFAULT_LEASE) —

    seconds a relay holds a claimed record

Raises:

  • (ArgumentError) 


140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/acemq/amqp/patterns/sql_outbox.rb', line 140

def initialize(connection:, relay: nil, table: DEFAULT_TABLE,
               max_attempts: DEFAULT_MAX_ATTEMPTS, lease: DEFAULT_LEASE)
  @source = connection
  @relay_source = relay || connection
  @table = SQL.table_name!(table)
  raise ArgumentError, "max_attempts must be at least 1" if max_attempts.to_i < 1
  raise ArgumentError, "a lease must be positive" unless lease.to_f.positive?

  @max_attempts = max_attempts.to_i
  @lease = lease.to_f
end

Instance Attribute Details

#table ⇒ Object (readonly)

The table this store reads and writes.



153
154
155
 
# File 'lib/acemq/amqp/patterns/sql_outbox.rb', line 153

def table
  @table
end

Class Method Details

.ddl(table) ⇒ Object

The table, as the two statements that make it.

Written here rather than read from a .sql file because the gem ships lib/**/*.rb and nothing else, and a schema that is missing from the package is a schema that works in the repository and nowhere after it. Keeping it as separate statements also avoids splitting a file on semicolons, which goes wrong the first time a comment contains one.



105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/acemq/amqp/patterns/sql_outbox.rb', line 105

def self.ddl(table)
  [
    <<~SQL.chomp,
      CREATE TABLE IF NOT EXISTS #{table} (
        id             VARCHAR(64)              NOT NULL PRIMARY KEY,
        exchange_name  VARCHAR(255)             NOT NULL,
        routing_key    VARCHAR(255)             NOT NULL,
        body           TEXT                     NOT NULL,
        body_encoding  VARCHAR(16)              NOT NULL,
        content_type   VARCHAR(255)             NOT NULL,
        headers        TEXT                     NOT NULL,
        created_at     TIMESTAMP WITH TIME ZONE NOT NULL,
        published_at   TIMESTAMP WITH TIME ZONE,
        attempts       INTEGER                  NOT NULL DEFAULT 0,
        last_error     VARCHAR(1000),
        locked_by      VARCHAR(64),
        locked_until   TIMESTAMP WITH TIME ZONE
      )
    SQL
    # The relay's only query is "oldest unpublished first", so that is
    # what is indexed.
    "CREATE INDEX IF NOT EXISTS #{table}_pending ON #{table} (published_at, created_at)"
  ]
end

Instance Method Details

#add(record, connection: nil) ⇒ Object

Records a message to be published, on the caller's connection.

Neither committed nor closed here. The connection belongs to the caller's transaction, and this insert becomes durable exactly when that transaction does — which is the guarantee the whole pattern rests on.

Adding the same record twice is not an error, because the caller may be retrying its own transaction, but it must not become two messages.

Parameters:

  • record (OutboxRecord)
  • connection (Object, nil) (defaults to: nil) —

    the connection the caller's transaction is on, when it is not the one this store was built with

Raises:

  • (ArgumentError) 


180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/acemq/amqp/patterns/sql_outbox.rb', line 180

def add(record, connection: nil)
  raise ArgumentError, "an outbox record needs an id" if record.id.to_s.empty?

  body, encoding = encoded(record.body)
  sql = "INSERT INTO #{@table} (#{COLUMNS}, published_at, locked_by, locked_until) " \
        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL)"
  # SQL.key rather than to_s: an id that came off the wire is
  # ASCII-8BIT, sqlite3 binds that as a blob, and a blob never matches
  # the text the same id was written as. Here that would turn "adding
  # the same record twice is not an error" into two rows and two
  # published messages. See SQL.key.
  run_unless_taken(sql, [SQL.key(record.id), record.exchange.to_s,
                         record.routing_key.to_s,
                         body, encoding, record.content_type.to_s,
                         JSON.generate(record.headers || {}),
                         SQL.at_utc(record.created_at || Time.now), 0, nil],
                   on: connection_for(connection))
  nil
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, alongside the business tables it commits with — a library that creates tables at start-up has taken a decision about when your database changes that is not its to take.



162
163
164
165
 
# File 'lib/acemq/amqp/patterns/sql_outbox.rb', line 162

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

#mark_failed(id, reason) ⇒ Object

Records that publishing a record failed, and gives up its lease.

Counting the attempt is what eventually stops a record nothing can publish from being claimed on every sweep for ever, and releasing the lease is what lets the next sweep try it rather than waiting out a minute for nothing.



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

def mark_failed(id, reason)
  run("UPDATE #{@table} SET attempts = attempts + 1, last_error = ?, " \
      "locked_by = NULL, locked_until = NULL WHERE id = ?",
      [truncate(reason), SQL.key(id)])
  nil
rescue StandardError
  # Swallowed on purpose. This is already the failure path, and a relay
  # that died here would leave the record leased rather than free;
  # letting the lease expire is the gentler outcome, and the message is
  # still in the table either way.
  nil
end

#mark_published(id) ⇒ Object

Removes a record the broker has confirmed.

The lease is cleared along with the mark, so the row reads plainly afterwards: a published row still showing a lock confuses whoever is reading the table during an incident.



228
229
230
231
232
 
# File 'lib/acemq/amqp/patterns/sql_outbox.rb', line 228

def mark_published(id)
  run("UPDATE #{@table} SET published_at = ?, locked_by = NULL, locked_until = NULL " \
      "WHERE id = ?", [SQL.at_utc(Time.now), SQL.key(id)])
  nil
end

#pending(limit = 0) ⇒ Array<OutboxRecord>

The records waiting, oldest first, claimed under a lease.

Claiming rather than merely reading, because two relays that both read the same batch both publish it, and the duplicate is the one thing a relay can avoid without any help from the consumer. What comes back is what this call won, which may be fewer than were asked for and may be none.

Parameters:

  • limit (Integer) (defaults to: 0) —

    how many at most; zero for the maximum batch

Returns:



210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/acemq/amqp/patterns/sql_outbox.rb', line 210

def pending(limit = 0)
  token = SecureRandom.uuid
  now = Time.now.utc
  candidates = candidate_ids(limit_of(limit), now)
  return [] if candidates.empty?
  # Every candidate went to another relay between the select and the
  # update. Not an error, and not worth a second round trip: the next
  # sweep will find more.
  return [] if take_lease(candidates, token, now).zero?

  claimed(token)
end

#pending_count ⇒ Object Also known as: size

How many records are waiting.



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

def pending_count
  result = run("SELECT COUNT(*) FROM #{@table} WHERE published_at IS NULL", [])
  result.rows.dig(0, 0).to_i
end

#purge_published(older_than) ⇒ Integer

Removes published records older than older_than seconds.

Nothing deletes them otherwise, and an outbox table that only grows eventually makes the relay's own query slow. How long to keep them is a judgement about auditing rather than about messaging, so it is the caller's to make.

Returns:

  • (Integer) —

    how many rows were removed



268
269
270
271
272
 
# File 'lib/acemq/amqp/patterns/sql_outbox.rb', line 268

def purge_published(older_than)
  cutoff = SQL.at_utc(Time.now - older_than.to_f)
  run("DELETE FROM #{@table} WHERE published_at IS NOT NULL AND published_at < ?",
      [cutoff]).affected.to_i
end

#to_s ⇒ Object 



274
 
# File 'lib/acemq/amqp/patterns/sql_outbox.rb', line 274

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