ID validation

This patch adds `auto_increment_increment` validation to the following
scenarios:
 * Connection initalization
 * Generating one ID
 * Generating multiple IDs

These changes should have no noticeable impact on performance and are
thread safe.

The intent is to prevent data corruption when a server is misconfigured
and in the event that one is found, it's removed from the pool of
available alloc servers allowing creation to continue with the valid
alloc servers.

Validation during initialization is done by comparing the client
configuration and with what's set on each server. If the server is
misconfigured, an error is reported via the `notifier` and it's left out
of the pool.

Validation during creation happens by comparing each pair of identifiers
and ensuring they've been incremented correctly by the configured amount.

If the `auto_increment_increment` changes while the clients are using
the server, the alloc server is removed from the list of servers and
an error is reported via the `notifier`.

An InvalidIncrementException is raised if all servers are misconfigured.

global_uid.gemspec

@@ -17,6 +17,7 @@ Gem::Specification.new 'global_uid', '3.7.1' do |s|
   s.add_development_dependency('minitest')
   s.add_development_dependency('minitest-rg')
   s.add_development_dependency('mocha')
+  s.add_development_dependency('benchmark-ips')
   s.add_development_dependency('bump')
   s.add_development_dependency('phenix')
 

lib/global_uid.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 require "global_uid/base"
+require "global_uid/allocations"
 require "global_uid/active_record_extension"
 require "global_uid/has_and_belongs_to_many_builder_extension"
 require "global_uid/migration_extension"
@@ -9,6 +10,8 @@ module GlobalUid
   class NoServersAvailableException < StandardError ; end
   class ConnectionTimeoutException < StandardError ; end
   class TimeoutException < StandardError ; end
+  class InvalidIncrementException < StandardError ; end
+  class InvalidIncrementPreventionException < StandardError ; end
 
   autoload :ServerVariables, "global_uid/server_variables"
 end

lib/global_uid/allocations.rb

@@ -0,0 +1,28 @@
+class Allocations
+  attr_reader :allocations, :max_size, :incrementing_by
+
+  def initialize(max_size: 5, incrementing_by:)
+    @allocations = []
+    @max_size = max_size
+    @incrementing_by = incrementing_by
+  end
+
+  def add(identifier)
+    allocations.shift if allocations.size >= max_size
+    allocations << identifier
+    valid_increment?
+  end
+  alias_method :<<, :add
+
+  def to_s
+    allocations.to_s
+  end
+
+  private
+
+  def valid_increment?
+    allocations[1..-1].all? do |identifier|
+      (identifier - allocations[0]) % incrementing_by == 0
+    end
+  end
+end

lib/global_uid/base.rb

@@ -9,7 +9,7 @@ class Base
     GLOBAL_UID_DEFAULTS = {
       :connection_timeout   => 3,
       :connection_retry     => 10.minutes,
-      :notifier             => Proc.new { |exception, message| ActiveRecord::Base.logger.error("GlobalUID error:  #{exception} #{message}") },
+      :notifier             => Proc.new { |exception, message| ActiveRecord::Base.logger.error("GlobalUID error: #{exception} #{message}") },
       :query_timeout        => 10,
       :increment_by         => 5,  # This will define the maximum number of servers that you can have
       :disabled             => false,
@@ -87,6 +87,7 @@ def self.init_server_info(options)
         info[:offset]   = offset
         info[:rand]     = rand
         info[:new?]     = true
+        info[:allocated_ids] = Allocations.new(incrementing_by: self.global_uid_increment_by)
         offset +=1
         info
       end
@@ -112,9 +113,18 @@ def self.setup_connections!(options)
         if info[:new?] || ( info[:retry_at] && Time.now > info[:retry_at] )
           info[:new?] = false
 
-          connection = new_connection(info[:name], connection_timeout, info[:offset], increment_by)
-          info[:cx]  = connection
-          info[:retry_at] = Time.now + options[:connection_retry] if connection.nil?
+          begin
+            connection = new_connection(info[:name], connection_timeout, info[:offset], increment_by)
+            if !connection.nil? && increment_by != (db_increment = connection.select_value("SELECT @@auto_increment_increment"))
+              raise InvalidIncrementPreventionException, "Configured: '#{increment_by}', Found: '#{db_increment}' on '#{connection.current_database}'"
+            end
+
+            info[:cx]  = connection
+            info[:retry_at] = Time.now + options[:connection_retry] if connection.nil?
+          rescue InvalidIncrementPreventionException => e
+            notify e, "#{e.message}"
+            info[:cx]  = nil
+          end
         end
       end
 
@@ -156,18 +166,27 @@ def self.with_connections(options = {})
 
     def self.notify(exception, message)
       if self.global_uid_options[:notifier]
-        self.global_uid_options[:notifier].call(exception, message)
+        self.global_uid_options[:notifier].call(exception.class, message)
       end
     end
 
     def self.get_connections(options = {})
       with_connections {}
     end
 
+    def self.validate_increment!(new_record_id, connection)
+      server = self.servers.find { |s| connection.current_database.include?(s[:name]) }
+
+      unless server[:allocated_ids] << new_record_id
+        raise InvalidIncrementException, "Recently allocated IDs: #{server[:allocated_ids]} on '#{connection.current_database}'"
+      end
+    end
+
     def self.get_uid_for_class(klass, options = {})
       with_connections do |connection|
         Timeout.timeout(self.global_uid_options[:query_timeout], TimeoutException) do
           id = connection.insert("REPLACE INTO #{klass.global_uid_table} (stub) VALUES ('a')")
+          validate_increment!(id, connection)
           return id
         end
       end
@@ -179,6 +198,9 @@ def self.get_many_uids_for_class(klass, count, options = {})
       with_connections do |connection|
         Timeout.timeout(self.global_uid_options[:query_timeout], TimeoutException) do
           increment_by = connection.select_value("SELECT @@auto_increment_increment")
+          if increment_by != self.global_uid_increment_by
+            raise InvalidIncrementPreventionException, "Configured: '#{self.global_uid_increment_by}', Found: '#{increment_by}' on '#{connection.current_database}'"
+          end
           start_id = connection.insert("REPLACE INTO #{klass.global_uid_table} (stub) VALUES " + (["('a')"] * count).join(','))
           return start_id.step(start_id + (count-1) * increment_by, increment_by).to_a
         end
@@ -198,6 +220,10 @@ def self.global_uid_servers
       self.global_uid_options[:id_servers]
     end
 
+    def self.global_uid_increment_by
+      self.global_uid_options[:increment_by]
+    end
+
     def self.id_table_from_name(name)
       "#{name}_ids".to_sym
     end

test/global_uid_test.rb

@@ -205,6 +205,33 @@ def table_exists?(connection, table)
     end
   end
 
+  describe "Performance" do
+    before do
+      CreateWithNoParams.up
+      CreateWithoutGlobalUIDs.up
+    end
+
+    it "has a negligible performance impact" do
+      report = Benchmark.ips do |x|
+        x.report(WithGlobalUID)    { WithGlobalUID.create! }
+        x.report(WithoutGlobalUID) { WithoutGlobalUID.create! }
+        x.compare!
+      end
+
+      with_global_uid    = report.entries.find { |e| e.label == WithGlobalUID }.stats
+      without_global_uid = report.entries.find { |e| e.label == WithoutGlobalUID }.stats
+
+      performance_impact, _ = with_global_uid.slowdown(without_global_uid)
+      # assert performance_impact < 2 # The results vary too much on CI :(
+    end
+
+    after do
+      reset_connections!
+      CreateWithNoParams.down
+      CreateWithoutGlobalUIDs.down
+    end
+  end
+
   describe "With GlobalUID" do
     before do
       CreateWithNoParams.up
@@ -222,6 +249,90 @@ def table_exists?(connection, table)
       it "get a unique id" do
         test_unique_ids
       end
+
+      describe 'when the auto_increment_increment changes' do
+        before do
+          @notifications = []
+          GlobalUid::Base.global_uid_options[:notifier] = Proc.new do |exception, message|
+            GlobalUid::Base::GLOBAL_UID_DEFAULTS[:notifier].call(exception, message)
+            @notifications << exception
+          end
+        end
+
+        describe "and all servers report a value other than what's configured" do
+          it "raises an exception when configuration incorrect during initialization" do
+            GlobalUid::Base.global_uid_options[:increment_by] = 42
+            reset_connections!
+            assert_raises(GlobalUid::NoServersAvailableException) { test_unique_ids(10) }
+            assert_includes(@notifications, GlobalUid::InvalidIncrementPreventionException)
+          end
+
+          it "raises an exception, preventing duplicate ID generation" do
+            GlobalUid::Base.with_connections do |con|
+              con.execute("SET SESSION auto_increment_increment = 42")
+            end
+
+            assert_raises(GlobalUid::NoServersAvailableException) { test_unique_ids(10) }
+            assert_includes(@notifications, GlobalUid::InvalidIncrementException)
+          end
+
+          it "raises an exception before attempting to generate many UIDs" do
+            GlobalUid::Base.with_connections do |con|
+              con.execute("SET SESSION auto_increment_increment = 42")
+            end
+
+            assert_raises GlobalUid::NoServersAvailableException do
+              WithGlobalUID.generate_many_uids(10)
+            end
+            assert_includes(@notifications, GlobalUid::InvalidIncrementPreventionException)
+          end
+
+          it "doesn't cater for increment_by being increased by a factor of x" do
+            GlobalUid::Base.with_connections do |connection|
+              connection.execute("SET SESSION auto_increment_increment = #{GlobalUid::Base::GLOBAL_UID_DEFAULTS[:increment_by] * 2}")
+            end
+            # Due to multiple processes and threads sharing the same alloc server, identifiers may be provisioned
+            # before the current thread receives its next one. We rely on the gap being divisible by the configured increment
+            test_unique_ids(10)
+            assert_empty(@notifications)
+          end
+        end
+
+        describe "and only one server reports a value other than what's configured" do
+          it "notifies the client when configuration incorrect during initialization" do
+            modified_connection = Proc.new do |name, connection_timeout, offset, increment_by|
+              config = ActiveRecord::Base.configurations.to_h[name]
+              ActiveRecord::Base.mysql2_connection(config).tap do |connection|
+                connection.execute("SET SESSION auto_increment_increment = 42") if name == "test_id_server_1"
+              end
+            end
+
+            GlobalUid::Base.stub :new_connection, modified_connection do
+              reset_connections!
+              test_unique_ids(32)
+              assert_includes(@notifications, GlobalUid::InvalidIncrementPreventionException)
+            end
+          end
+
+          it "notifies the client and continues with the other connection" do
+            con = GlobalUid::Base.get_connections.first
+            con.execute("SET SESSION auto_increment_increment = 42")
+
+            # Trigger the exception, one call may not hit the server, there's still a 1/(2^32) chance of failure.
+            test_unique_ids(32)
+            assert_includes(@notifications, GlobalUid::InvalidIncrementException)
+          end
+
+          it "notifies the client and continues when attempting to generate many UIDs" do
+            con = GlobalUid::Base.get_connections.first
+            con.execute("SET SESSION auto_increment_increment = 42")
+
+            # Trigger the exception, one call may not hit the server, there's still a 1/(2^32) chance of failure.
+            32.times { WithGlobalUID.generate_many_uids(10) }
+            assert_includes(@notifications, GlobalUid::InvalidIncrementPreventionException)
+          end
+        end
+      end
     end
 
     describe "With a timing out server" do
@@ -422,15 +533,15 @@ def parent_child_fork_values
 
   private
 
-  def test_unique_ids
+  def test_unique_ids(amount = 10)
     seen = {}
-    (0..10).each do
-      foo = WithGlobalUID.new
-      foo.save
-      refute_nil foo.id
-      assert_nil foo.description
-      refute seen.has_key?(foo.id)
-      seen[foo.id] = 1
+    amount.times do
+      WithGlobalUID.create!.tap do |record|
+        refute_nil record.id
+        assert_nil record.description
+        refute seen.has_key?(record.id)
+        seen[record.id] = true
+      end
     end
   end
 
@@ -439,9 +550,13 @@ def reset_connections!
   end
 
   def restore_defaults!
-    GlobalUid::Base.global_uid_options[:storage_engine] = nil
-    GlobalUid::Base.global_uid_options[:disabled] = false
-    GlobalUid::Base.global_uid_options[:dry_run] = false
+    GlobalUid::Base.global_uid_options = {
+      :disabled   => false,
+      :id_servers => [
+        "test_id_server_1",
+        "test_id_server_2"
+      ]
+    }
   end
 
   def show_create_sql(klass, table)

test/models.rb

@@ -3,6 +3,7 @@ class WithGlobalUID < ActiveRecord::Base
 end
 
 class WithoutGlobalUID < ActiveRecord::Base
+  disable_global_uid
 end
 
 class Parent < ActiveRecord::Base

test/test_helper.rb

@@ -1,19 +1,13 @@
 # frozen_string_literal: true
 require 'bundler/setup'
 require "active_record"
+require 'benchmark/ips'
 require 'minitest/autorun'
 require 'minitest/rg'
 require 'mocha/setup'
 require 'global_uid'
 require 'phenix'
 
-GlobalUid::Base.global_uid_options = {
-  :disabled   => false,
-  :id_servers => [
-    "test_id_server_1",
-    "test_id_server_2"
-  ]
-}
 GlobalUid::Base.extend(GlobalUid::ServerVariables)
 
 Phenix.configure do |config|