Class: AceMQ::AMQP::Patterns::SQLSchemaRegistry

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

Overview

A schema registry in a database table.

registry = Patterns::SQLSchemaRegistry.new(connection: db)
registry.create_schema                     # development and tests only

schema = registry.register("order.placed", "avro", definition)
registry.by_id(schema.id)

InMemorySchemaRegistry is fine for a test and useless for anything else: identifiers must be stable for ever, because a message published today may be read next year by a consumer looking up the shape it was written with. A registry that hands out fresh identifiers on restart makes every message written before the restart unreadable, and does it silently.

This is the smallest thing that fixes that. It is not Confluent's registry and does not try to be: no compatibility checking, no versioning interface, no HTTP. What it does is remember which integer stands for which schema, across restarts and across processes.

Answers are cached in memory and never invalidated, because neither can change: an identifier stands for one schema for ever, and a schema keeps the identifier it was given.

As with InMemorySchemaRegistry, nothing here puts anything on the wire. Which header carries a schema identifier is a cross-language contract and not one AceMQ has agreed yet.

Constant Summary collapse

DEFAULT_TABLE = 
"acemq_schema_registry"
REGISTRATION_ATTEMPTS =

Two is enough to settle the fingerprint race; the rest are for a database having a day.

5
COLUMNS =

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

"id, subject, schema_version, format, definition, fingerprint, registered_at"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection:, table: DEFAULT_TABLE) ⇒ SQLSchemaRegistry

Returns a new instance of SQLSchemaRegistry.

Parameters:

  • connection (Object, #call) —

    where the table lives

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

    a plain SQL identifier



110
111
112
113
114
115
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 110

def initialize(connection:, table: DEFAULT_TABLE)
  @source = connection
  @table = SQL.table_name!(table)
  @lock = Mutex.new
  @by_id = {}
end

Instance Attribute Details

#table ⇒ Object (readonly)

The table this registry reads and writes.



118
119
120
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 118

def table
  @table
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.

TEXT rather than the CLOB the Java schema uses: PostgreSQL has no CLOB, and a definition is text everywhere that does.



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 69

def self.ddl(table)
  [
    <<~SQL.chomp,
      CREATE TABLE IF NOT EXISTS #{table} (
        id            INTEGER                  NOT NULL PRIMARY KEY,
        subject       VARCHAR(255)             NOT NULL,
        schema_version INTEGER                 NOT NULL,
        format        VARCHAR(32)              NOT NULL,
        definition    TEXT                     NOT NULL,
        fingerprint   VARCHAR(64)              NOT NULL,
        registered_at TIMESTAMP WITH TIME ZONE NOT NULL
      )
    SQL
    # The fingerprint is what makes registration idempotent: the same
    # definition offered twice must come back with the same id, from any
    # process, for ever. Unique rather than merely indexed, so two
    # processes racing to register the same schema end with one row and
    # one loser that re-reads instead of a second id for the same bytes.
    "CREATE UNIQUE INDEX IF NOT EXISTS #{table}_fingerprint " \
    "ON #{table} (subject, fingerprint)",
    # One row, holding the last identifier handed out.
    #
    # A sequence would be the obvious tool and is spelled differently on
    # every database this has to run on. Taking the highest id and
    # adding one needs no sequence and is wrong under load: two writers
    # registering two different schemas at the same moment compute the
    # same next id, and one of them loses a race it cannot win by
    # retrying, because the writer it lost to is doing the same
    # arithmetic. Updating this row takes a row lock, so writers queue
    # for an instant and every one of them gets a number.
    <<~SQL.chomp
      CREATE TABLE IF NOT EXISTS #{table}_seq (
        only_row INTEGER NOT NULL PRIMARY KEY,
        last_id  INTEGER NOT NULL
      )
    SQL
  ]
end

Instance Method Details

#by_id(id) ⇒ Object

The schema an identifier names.

Raises:



169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 169

def by_id(id)
  cached = @lock.synchronize { @by_id[id.to_i] }
  return cached if cached

  rows = run("SELECT #{COLUMNS} FROM #{@table} WHERE id = ?", [id.to_i]).rows
  if rows.empty?
    # A message naming a schema this registry has never held is usually
    # one environment reading another's messages, which is worth saying
    # plainly: the alternative guess, a corrupt message, sends people
    # looking in the wrong place.
    raise SchemaNotFound,
          "no schema with id #{id}. Either a message was written against a schema " \
          "registered somewhere else, or this is not the registry that wrote it."
  end

  remember(schema_from(rows[0]))
end

#create_schema ⇒ Object

Creates the tables if they are not already there, and seeds the counter so the first registration has something to lock rather than something to create.



123
124
125
126
127
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 123

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

#latest(subject) ⇒ Object

The newest version of a subject.

Raises:



190
191
192
193
194
195
196
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 190

def latest(subject)
  rows = run("SELECT #{COLUMNS} FROM #{@table} WHERE subject = ? " \
             "ORDER BY schema_version DESC LIMIT 1", [subject.to_s]).rows
  raise SchemaNotFound, "no schema for subject #{subject.inspect}" if rows.empty?

  remember(schema_from(rows[0]))
end

#register(subject, format, definition) ⇒ SchemaDefinition

Records a schema and returns it with an identifier.

The same definition registered twice returns the same identifier rather than making a second version. Without that, a service that registers its schemas on every start adds a version per restart, and a week later the subject has three hundred identical versions.

Parameters:

  • subject (String) —

    groups the versions of one message type

  • format (String) —

    "avro", "protobuf", "json-schema" — nothing here interprets it

  • definition (String) —

    the schema

Returns:

Raises:



141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 141

def register(subject, format, definition)
  if subject.to_s.empty? || definition.to_s.empty?
    raise ArgumentError, "a schema needs a subject and a definition"
  end

  fingerprint = Patterns.fingerprint(definition)
  # The loop is for one race: two writers registering the same schema at
  # the same moment. The unique index lets exactly one of them insert,
  # and the loser comes back round to read the winner's row rather than
  # inventing a second identifier for the same bytes.
  REGISTRATION_ATTEMPTS.times do
    found = by_fingerprint(subject.to_s, fingerprint)
    return remember(found) if found
    next unless insert_new(subject.to_s, format.to_s, definition.to_s, fingerprint)

    return remember(by_fingerprint(subject.to_s, fingerprint))
  end

  raise FatalError,
        "could not register #{subject.inspect} after #{REGISTRATION_ATTEMPTS} " \
        "attempts. Every insert was refused for breaking a constraint and no " \
        "matching fingerprint was there afterwards, so #{@table} has a constraint " \
        "this registry does not know about -- check what a migration added to it."
end

#size ⇒ Object

How many schemas this registry holds.



209
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 209

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

#to_s ⇒ Object 



211
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 211

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

#versions(subject) ⇒ Object

Every version of a subject, oldest first.

Empty when there are none, because "what versions are there" has a sensible answer for a subject nobody has registered and "which schema is this" does not.



203
204
205
206
 
# File 'lib/acemq/amqp/patterns/sql_schema.rb', line 203

def versions(subject)
  run("SELECT #{COLUMNS} FROM #{@table} WHERE subject = ? ORDER BY schema_version",
      [subject.to_s]).rows.map { |row| remember(schema_from(row)) }
end