optimistic locking version

lib/mysql_session.rb

@@ -15,12 +15,13 @@ class ActiveRecord::ConnectionAdapters::MysqlAdapter
 
 class MysqlSession
 
-  attr_accessor :id, :session_id, :data
+  attr_accessor :id, :session_id, :data, :lock_version
 
   def initialize(session_id, data)
     @session_id = session_id
     @data = data
     @id = nil
+    @lock_version = 0
   end
 
   class << self
@@ -44,21 +45,32 @@ def query(sql)
     # +created_at+ and +updated_at+ as they are not accessed anywhyere
     # outside this class
     def find_session(session_id, lock = false)
+      find("`session_id`='#{Mysql.quote session_id}' LIMIT 1" + (lock ? ' FOR UPDATE' : ''))
+    end
+
+    def find_by_primary_id(primary_key_id, lock = false)
+      if primary_key_id
+        find("`id`='#{primary_key_id}'" + (lock ? ' FOR UPDATE' : ''))
+      else
+        nil
+      end
+    end
+
+    def find(conditions)
       connection = session_connection
       connection.query_with_result = true
-      session_id = Mysql::quote(session_id)
-      result = query("SELECT id, data FROM sessions WHERE `session_id`='#{session_id}' LIMIT 1" + (lock ? ' FOR UPDATE' : ''))
-      my_session = nil
+      result = query("SELECT session_id, data,id #{  SqlSession.locking_enabled? ? ',lock_version ' : ''} FROM sessions WHERE " + conditions)
+       my_session = nil
       # each is used below, as other methods barf on my 64bit linux machine
       # I suspect this to be a bug in mysql-ruby
       result.each do |row|
-        my_session = new(session_id, row[1])
-        my_session.id = row[0]
+        my_session = new(row[0], row[1])
+        my_session.id = row[2]
+        my_session.lock_version = row[3].to_i
       end
       result.free
       my_session
-    end
-
+    end  
     # create a new session with given +session_id+ and +data+
     # and save it immediately to the database
     def create_session(session_id, data)
@@ -87,15 +99,32 @@ def update_session(data)
     if @id
       # if @id is not nil, this is a session already stored in the database
       # update the relevant field using @id as key
-      self.class.query("UPDATE sessions SET `updated_at`=NOW(), `data`='#{Mysql::quote(data)}' WHERE id=#{@id}")
+      if SqlSession.locking_enabled?
+        self.class.query("UPDATE sessions SET `updated_at`=NOW(), `data`='#{Mysql::quote(data)}', lock_version=lock_version+1 WHERE id=#{@id}")
+        @lock_version += 1 #if we are here then we hold a lock on the table - we know our version is up to date
+      else
+        self.class.query("UPDATE sessions SET `updated_at`=NOW(), `data`='#{Mysql::quote(data)}' WHERE id=#{@id}")
+      end
     else
       # if @id is nil, we need to create a new session in the database
       # and set @id to the primary key of the inserted record
       self.class.query("INSERT INTO sessions (`updated_at`, `session_id`, `data`) VALUES (NOW(), '#{@session_id}', '#{Mysql::quote(data)}')")
       @id = connection.insert_id
+      @lock_version = 0
     end
   end
 
+  def update_session_optimistically(data)
+    raise 'cannot update unsaved record optimistically' unless @id
+    connection = self.class.session_connection
+    self.class.query("UPDATE sessions SET `updated_at`=NOW(), `data`='#{Mysql::quote(data)}', `lock_version`=`lock_version`+1 WHERE id=#{@id} AND lock_version=#{@lock_version}")
+    if connection.affected_rows == 1
+      @lock_version += 1
+      true
+    else
+      false
+    end
+  end
   # destroy the current session
   def destroy
     self.class.delete_all("session_id='#{session_id}'")

lib/postgresql_session.rb

@@ -51,6 +51,24 @@ def find_session(session_id, lock = false)
       my_session
     end
 
+    def find_by_primary_id(primary_key_id, lock = false)
+      if primary_key_id
+        connection = session_connection
+        quoted_session_id = connection.quote(session_id)
+        result = connection.query("SELECT session_id, data FROM sessions WHERE id=#{primary_key_id} LIMIT 1"  + (lock ? ' FOR UPDATE' : '') )
+        my_session = nil
+
+        if result[0] && result[0].size == 2
+          my_session = new(result[0][0], result[0][1])
+          my_session.id =primary_key_id
+        end
+
+        result.clear
+        my_session
+      else
+        nil
+      end
+    end
     # create a new session with given +session_id+ and +data+
     # and save it immediately to the database
     def create_session(session_id, data)

lib/session_smarts.rb

@@ -17,17 +17,63 @@ def save_session(session, data)
 
     return nil if changed_keys.empty? && deleted_keys.empty?
 
-    SqlSession.transaction do
-      fresh_session = session_class.find_session(session.session_id, true)
-      if fresh_session && fresh_session.data != original_marshalled_data && fresh_data = unmarshalize(fresh_session.data)
+    if SqlSession.locking_enabled?
+      begin
+        if session.id
+          while !session.update_session_optimistically(marshalize(data))        
+            fresh_session = get_fresh_session session, false
+            session,data = merge_sessions fresh_session, session, original_marshalled_data, changed_keys, deleted_keys, data
+          end
+        else
+          session.update_session(marshalize(data))
+        end
+      rescue ActiveRecord::StatementInvalid => e
+        if e.message =~ /Duplicate entry/
+          fresh_session = get_fresh_session session, false
+          session,data = merge_sessions fresh_session, session, original_marshalled_data, changed_keys, deleted_keys, data
+          retry
+        end
+        raise
+      end  
+    else
+      begin
+        SqlSession.transaction do
+          fresh_session = get_fresh_session session, true
+          session, data = merge_sessions fresh_session, session, original_marshalled_data, changed_keys, deleted_keys, data
+          session.update_session(marshalize(data))
+        end
+      rescue ActiveRecord::StatementInvalid => e
+        if e.message =~ /Duplicate entry/
+          retry
+        end
+        raise
+      end
+    end  
+
+
+    return data, session
+  end
+
+
+  def get_fresh_session session, lock
+    if session.id
+      fresh_session = session_class.find_by_primary_id session.id, lock
+    else
+      fresh_session = session_class.find_session session.session_id, false
+    end
+  end
+
+  def merge_sessions fresh_session, session, original_marshalled_data, changed_keys, deleted_keys, data
+    if fresh_session
+      data_changed = fresh_session.data != original_marshalled_data
+      if (data_changed || SqlSession.locking_enabled?) && fresh_data = unmarshalize(fresh_session.data)
         deleted_keys.each {|k| fresh_data.delete k}
         changed_keys.each {|k| fresh_data[k] = data[k]}
         data = fresh_data
         session = fresh_session
       end
-      session.update_session(marshalize(data))
     end
-    return data, session
+    return session, data
   end
 end
 

lib/sql_session.rb

@@ -26,4 +26,22 @@ def self.create_session(session_id, data)
   def update_session(data)
     update_attribute('data', data)
   end
+
+  # update session data using optimistic locking - return true on success, false if the record was stale
+  def update_session_optimistically data
+    update_attribute('data', data)
+    true
+  rescue ActiveRecord::StaleObjectError
+    false
+  end
+
+  #find the session record by its primary key id as opposed to its session id
+  def self.find_by_primary_id(primary_key_id, lock=false)
+    if primary_key_id
+      find primary_key_id, :lock => lock
+    else
+      nil
+    end
+  end
+
 end

lib/sqlite_session.rb

@@ -48,6 +48,23 @@ def find_session(session_id, lock = false)
       my_session
     end
 
+    def find_by_primary_id(primary_key_id, lock = false)
+      if primary_key_id
+        connection = session_connection
+        result = connection.execute("SELECT session_id, data FROM sessions WHERE `id`='#{primary_key_id}'")
+        my_session = nil
+        # each is used below, as other methods barf on my 64bit linux machine
+        # I suspect this to be a bug in mysql-ruby
+        result.each do |row|
+          my_session = new(row[0], row[1])
+          my_session.id = primary_key_id
+        end
+        my_session
+      else
+        nil
+      end
+    end
+
     # create a new session with given +session_id+ and +data+
     # and save it immediately to the database
     def create_session(session_id, data)

test/schema.rb

@@ -4,6 +4,7 @@
     t.column :data,       :text
     t.column :created_at, :timestamp
     t.column :updated_at, :timestamp
+    t.column :lock_version, :integer, :default => 0
   end
-  add_index :sessions, :session_id, :name => 'session_id_idx'
+  add_index :sessions, :session_id, :name => 'session_id_idx', :unique => true
 end

test/test_helper.rb

@@ -1,20 +1,14 @@
 $KCODE = 'u'
 require 'jcode'
 ENV['TZ'] = 'UTC'
-require 'rubygems'
-
-RAILS_VERSION='2.3.2' #the version of rails the tests run against
-gem 'activerecord', RAILS_VERSION
-gem 'activesupport', RAILS_VERSION
-gem 'actionpack', RAILS_VERSION
-
-RAILS_ENV = 'test'
-
 require 'test/unit'
+require 'rubygems'
+require '../rails_version.rb'
 require 'active_support'
 require 'active_record'
-require 'action_pack'
-require 'action_controller'
+require 'active_record/fixtures'
+require 'mocha'
+RAILS_ENV = 'test'
 
 require 'active_support/test_case'
 require 'active_record/fixtures'
@@ -27,6 +21,13 @@ class ActiveSupport::TestCase
     self.fixture_path = File.dirname(__FILE__) + "/fixtures/"
     self.use_instantiated_fixtures  = false
     self.use_transactional_fixtures = true
+
+    def with_locking
+      SqlSession.lock_optimistically = true
+      yield
+    ensure
+      SqlSession.lock_optimistically = false
+    end
   end
 
   def create_fixtures(*table_names, &block)
@@ -54,3 +55,5 @@ def create_fixtures(*table_names, &block)
 end
 
 load(File.dirname(__FILE__) + "/schema.rb")
+SqlSession.lock_optimistically = false
+