auto_increment_increment validation

Changelog.md

@@ -6,6 +6,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## [Unreleased]
 
+### Added
+- [Breaking change] ID Validation, ensure the ID coming back has been incremented using the configured `auto_increment_increment`. (https://github.com/zendesk/global_uid/pull/63)
+
 ### Removed
 - Removed the `dry_run` option (https://github.com/zendesk/global_uid/pull/64)
 - Removed `GlobalUid::ServerVariables` module (https://github.com/zendesk/global_uid/pull/66)

README.md

@@ -49,14 +49,15 @@ The `increment_by` value configured here does not dictate the value on your allo
 
 Here's a complete list of the options you can use:
 
-| Name                  | Default                                    | Description                                                                                                |
-| --------------------- | ------------------------------------------ | ---------------------------------------------------------------------------------------------------------- |
-| `:disabled`           | `false`                                    | Disable GlobalUid entirely                                                                                 |
-| `:connection_timeout` | 3 seconds                                  | Timeout for connecting to a global UID server                                                              |
-| `:query_timeout`      | 10 seconds                                 | Timeout for retrieving a global UID from a server before we move on to the next server                     |
-| `:connection_retry`   | 10 minutes                                 | After failing to connect or query a UID server, how long before we retry                                   |
-| `:notifier`           | A proc calling `ActiveRecord::Base.logger` | This proc is called with two parameters upon UID server failure -- an exception and a message              |
-| `:increment_by`       | 5                                          | Used to validate number of ID servers, preventing connections if there are more servers than the given increment |
+| Name                             | Default                                    | Description                                                                                                  |
+| -------------------------------- | ------------------------------------------ | ------------------------------------------------------------------------------------------------------------ |
+| `:disabled`                      | `false`                                    | Disable GlobalUid entirely                                                                                   |
+| `:connection_timeout`            | 3 seconds                                  | Timeout for connecting to a global UID server                                                                |
+| `:query_timeout`                 | 10 seconds                                 | Timeout for retrieving a global UID from a server before we move on to the next server                       |
+| `:connection_retry`              | 10 minutes                                 | After failing to connect or query a UID server, how long before we retry                                     |
+| `:notifier`                      | A proc calling `ActiveRecord::Base.logger` | This proc is called with two parameters upon UID server failure -- an exception and a message                |
+| `:increment_by`                  | 5                                          | Used for validation, compared with the value on the alloc servers to prevent allocation of duplicate IDs     |
+| `:suppress_increment_exceptions` | `false`                                    | Suppress configuration validation, allowing updates to `auto_increment_increment` while alloc servers in use |
 
 ### Migration
 

lib/global_uid.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 require "global_uid/base"
+require "global_uid/allocator"
 require "global_uid/active_record_extension"
 require "global_uid/has_and_belongs_to_many_builder_extension"
 require "global_uid/migration_extension"
@@ -9,6 +10,7 @@ module GlobalUid
   class NoServersAvailableException < StandardError ; end
   class ConnectionTimeoutException < StandardError ; end
   class TimeoutException < StandardError ; end
+  class InvalidIncrementException < StandardError ; end
 end
 
 ActiveRecord::Base.send(:include, GlobalUid::ActiveRecordExtension)

lib/global_uid/allocator.rb

@@ -0,0 +1,67 @@
+module GlobalUid
+  class Allocator
+    attr_reader :recent_allocations, :max_window_size, :incrementing_by, :connection
+
+    def initialize(incrementing_by:, connection:)
+      @recent_allocations = []
+      @max_window_size = 5
+      @incrementing_by = incrementing_by
+      @connection = connection
+      validate_connection_increment
+    end
+
+    def allocate_one(table)
+      identifier = connection.insert("REPLACE INTO #{table} (stub) VALUES ('a')")
+      allocate(identifier)
+    end
+
+    def allocate_many(table, count:)
+      increment_by = validate_connection_increment
+
+      start_id = connection.insert("REPLACE INTO #{table} (stub) VALUES " + (["('a')"] * count).join(','))
+      identifiers = start_id.step(start_id + (count - 1) * increment_by, increment_by).to_a
+      identifiers.each { |identifier| allocate(identifier) }
+      identifiers
+    end
+
+    private
+
+    def allocate(identifier)
+      recent_allocations.shift if recent_allocations.size >= max_window_size
+      recent_allocations << identifier
+
+      if !valid_allocation?
+        db_increment = connection.select_value("SELECT @@auto_increment_increment")
+        message = "Configured: '#{incrementing_by}', Found: '#{db_increment}' on '#{connection.current_database}'. Recently allocated IDs: #{recent_allocations}"
+        alert(InvalidIncrementException.new(message))
+      end
+
+      identifier
+    end
+
+    def valid_allocation?
+      recent_allocations[1..-1].all? do |identifier|
+        (identifier > recent_allocations[0]) &&
+          (identifier - recent_allocations[0]) % incrementing_by == 0
+      end
+    end
+
+    def validate_connection_increment
+      db_increment = connection.select_value("SELECT @@auto_increment_increment")
+
+      if db_increment != incrementing_by
+        alert(InvalidIncrementException.new("Configured: '#{incrementing_by}', Found: '#{db_increment}' on '#{connection.current_database}'"))
+      end
+
+      db_increment
+    end
+
+    def alert(exception)
+      if GlobalUid::Base.global_uid_options[:suppress_increment_exceptions]
+        GlobalUid::Base.notify(exception, exception.message)
+      else
+        raise exception
+      end
+    end
+  end
+end

lib/global_uid/base.rb

@@ -11,9 +11,10 @@ class Base
       :connection_retry     => 10.minutes,
       :notifier             => Proc.new { |exception, message| ActiveRecord::Base.logger.error("GlobalUID error: #{exception.class} #{message}") },
       :query_timeout        => 10,
-      :increment_by         => 5,  # This will define the maximum number of servers that you can have
+      :increment_by         => 5, # This will define the maximum number of servers that you can have
       :disabled             => false,
-      :per_process_affinity => true
+      :per_process_affinity => true,
+      :suppress_increment_exceptions => false
     }
 
     def self.servers
@@ -94,6 +95,7 @@ def self.disconnect!
 
     def self.setup_connections!
       connection_timeout = self.global_uid_options[:connection_timeout]
+      increment_by = self.global_uid_options[:increment_by]
 
       if self.servers.nil?
         self.servers = init_server_info
@@ -107,9 +109,18 @@ def self.setup_connections!
         if info[:new?] || ( info[:retry_at] && Time.now > info[:retry_at] )
           info[:new?] = false
 
-          connection = new_connection(info[:name], connection_timeout)
-          info[:cx]  = connection
-          info[:retry_at] = Time.now + self.global_uid_options[:connection_retry] if connection.nil?
+          begin
+            connection = new_connection(info[:name], connection_timeout)
+            info[:cx] = connection
+            if connection.nil?
+              info[:retry_at] = Time.now + self.global_uid_options[:connection_retry]
+            else
+              info[:allocator] = Allocator.new(incrementing_by: increment_by, connection: connection)
+            end
+          rescue InvalidIncrementException => e
+            notify e, "#{e.message}"
+            info[:cx] = nil
+          end
         end
       end
 
@@ -160,9 +171,9 @@ def self.get_connections
 
     def self.get_uid_for_class(klass)
       with_connections do |connection|
+        server = self.servers.find { |s| connection.current_database.include?(s[:name]) }
         Timeout.timeout(self.global_uid_options[:query_timeout], TimeoutException) do
-          id = connection.insert("REPLACE INTO #{klass.global_uid_table} (stub) VALUES ('a')")
-          return id
+          return server[:allocator].allocate_one(klass.global_uid_table)
         end
       end
       raise NoServersAvailableException, "All global UID servers are gone!"
@@ -171,10 +182,9 @@ def self.get_uid_for_class(klass)
     def self.get_many_uids_for_class(klass, count)
       return [] unless count > 0
       with_connections do |connection|
+        server = self.servers.find { |s| connection.current_database.include?(s[:name]) }
         Timeout.timeout(self.global_uid_options[:query_timeout], TimeoutException) do
-          increment_by = connection.select_value("SELECT @@auto_increment_increment")
-          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
+          return server[:allocator].allocate_many(klass.global_uid_table, count: count)
         end
       end
       raise NoServersAvailableException, "All global UID servers are gone!"

test/lib/allocator_test.rb

@@ -0,0 +1,96 @@
+# frozen_string_literal: true
+require_relative '../test_helper'
+
+describe GlobalUid::Allocator do
+  let(:connection)   { mock('connection') }
+  let(:increment_by) { 10 }
+  let(:allocator)    { GlobalUid::Allocator.new(incrementing_by: increment_by, connection: connection) }
+  let(:table)        { mock('table_class').class }
+
+  before do
+    restore_defaults!
+    connection.stubs(:current_database).returns('database_name')
+  end
+
+  describe 'with a configured increment_by that differs from the connections auto_increment_increment' do
+    it 'raises an exception to prevent erroneous ID allocation' do
+      connection.stubs(:select_value).with('SELECT @@auto_increment_increment').returns(50)
+      exception = assert_raises(GlobalUid::InvalidIncrementException) do
+        GlobalUid::Allocator.new(incrementing_by: 10, connection: connection)
+      end
+
+      assert_equal("Configured: '10', Found: '50' on 'database_name'", exception.message)
+    end
+  end
+
+  describe 'with a configured increment_by that matches the connections auto_increment_increment' do
+    before do
+      connection.stubs(:select_value).with('SELECT @@auto_increment_increment').returns(increment_by)
+    end
+
+    describe '#allocate_one' do
+      it 'allocates IDs, maintaining a small rolling selection of IDs for comparison' do
+        [10, 20, 30, 40, 50, 60, 70, 80].each do |id|
+          connection.expects(:insert).returns(id)
+          allocator.allocate_one(table)
+        end
+
+        assert_equal(5, allocator.max_window_size)
+        assert_equal([40, 50, 60, 70, 80], allocator.recent_allocations)
+      end
+
+      describe 'gap between ID not divisible by increment_by' do
+        it 'raises an error' do
+          connection.expects(:insert).returns(20)
+          allocator.allocate_one(table)
+
+          connection.stubs(:select_value).with('SELECT @@auto_increment_increment').returns(5)
+          connection.expects(:insert).returns(25)
+          exception = assert_raises(GlobalUid::InvalidIncrementException) do
+            allocator.allocate_one(table)
+          end
+
+          assert_equal("Configured: '10', Found: '5' on 'database_name'. Recently allocated IDs: [20, 25]", exception.message)
+        end
+      end
+
+      describe 'ID value does not increment upwards' do
+        it 'raises an error' do
+          connection.expects(:insert).returns(20)
+          allocator.allocate_one(table)
+
+          connection.expects(:insert).returns(20 - increment_by)
+          exception = assert_raises(GlobalUid::InvalidIncrementException) do
+            allocator.allocate_one(table)
+          end
+
+          assert_equal("Configured: '10', Found: '10' on 'database_name'. Recently allocated IDs: [20, 10]", exception.message)
+        end
+      end
+    end
+
+    describe '#allocate_many' do
+      it 'allocates IDs, maintaining a small rolling selection of IDs for comparison' do
+        connection.expects(:insert)
+          .with("REPLACE INTO Mocha::Mock (stub) VALUES ('a'),('a'),('a'),('a'),('a'),('a'),('a'),('a')")
+          .returns(10)
+        allocator.allocate_many(table, count: 8)
+
+        assert_equal(5, allocator.max_window_size)
+        assert_equal([40, 50, 60, 70, 80], allocator.recent_allocations)
+      end
+
+      describe 'with a configured increment_by that differs from the active connection auto_increment_increment' do
+        it 'raises an error' do
+          connection.stubs(:select_value).with('SELECT @@auto_increment_increment').returns(5)
+          connection.expects(:insert).never
+          exception = assert_raises(GlobalUid::InvalidIncrementException) do
+            allocator.allocate_many(table, count: 8)
+          end
+
+          assert_equal("Configured: '10', Found: '5' on 'database_name'", exception.message)
+        end
+      end
+    end
+  end
+end