Merge pull request #27 from crystal-lang/feature/nested-transactions

Feature/nested transactions

spec/dummy_driver.cr

@@ -42,11 +42,57 @@ class DummyDriver < DB::Driver
       @connected = false
     end
 
+    def create_transaction
+      DummyTransaction.new(self)
+    end
+
     protected def do_close
       super
     end
   end
 
+  class DummyTransaction < DB::TopLevelTransaction
+    getter committed = false
+    getter rolledback = false
+
+    def initialize(connection)
+      super(connection)
+    end
+
+    def commit
+      super
+      @committed = true
+    end
+
+    def rollback
+      super
+      @rolledback = true
+    end
+
+    protected def create_save_point_transaction(parent, savepoint_name : String)
+      DummySavePointTransaction.new(parent, savepoint_name)
+    end
+  end
+
+  class DummySavePointTransaction < DB::SavePointTransaction
+    getter committed = false
+    getter rolledback = false
+
+    def initialize(parent, savepoint_name)
+      super(parent, savepoint_name)
+    end
+
+    def commit
+      super
+      @committed = true
+    end
+
+    def rollback
+      super
+      @rolledback = true
+    end
+  end
+
   class DummyStatement < DB::Statement
     property params
 

spec/save_point_transaction_spec.cr

@@ -0,0 +1,160 @@
+require "./spec_helper"
+
+private class FooException < Exception
+end
+
+private def with_dummy_top_transaction
+  with_dummy_connection do |cnn|
+    cnn.transaction do |tx|
+      yield tx.as(DummyDriver::DummyTransaction), cnn
+    end
+  end
+end
+
+private def with_dummy_nested_transaction
+  with_dummy_connection do |cnn|
+    cnn.transaction do |tx|
+      tx.transaction do |nested|
+        yield nested.as(DummyDriver::DummySavePointTransaction), cnn
+      end
+    end
+  end
+end
+
+describe DB::SavePointTransaction do
+  {% for context in [:with_dummy_top_transaction, :with_dummy_nested_transaction] %}
+  describe "{{context.id}}" do
+    it "begin/commit transaction from parent transaction" do
+      {{context.id}} do |parent_tx|
+        tx = parent_tx.begin_transaction
+        tx.commit
+      end
+    end
+
+    it "begin/rollback transaction from parent transaction" do
+      {{context.id}} do |parent_tx|
+        tx = parent_tx.begin_transaction
+        tx.rollback
+      end
+    end
+
+    it "raise if begin over existing transaction" do
+      {{context.id}} do |parent_tx|
+        parent_tx.begin_transaction
+        expect_raises(DB::Error, "There is an existing nested transaction in this transaction") do
+          parent_tx.begin_transaction
+        end
+      end
+    end
+
+    it "allow sequential transactions" do
+      {{context.id}} do |parent_tx|
+        tx = parent_tx.begin_transaction
+        tx.rollback
+
+        tx = parent_tx.begin_transaction
+        tx.commit
+      end
+    end
+
+    it "transaction with block from parent transaction should be committed" do
+      t = uninitialized DummyDriver::DummySavePointTransaction
+
+      with_witness do |w|
+        {{context.id}} do |parent_tx|
+          parent_tx.transaction do |tx|
+            if tx.is_a?(DummyDriver::DummySavePointTransaction)
+              t = tx
+              w.check
+            end
+          end
+        end
+      end
+
+      t.committed.should be_true
+      t.rolledback.should be_false
+    end
+  end
+  {% end %}
+
+  it "only nested transaction with block from parent transaction should be rolledback if raise DB::Rollback" do
+    top = uninitialized DummyDriver::DummyTransaction
+    t = uninitialized DummyDriver::DummySavePointTransaction
+
+    with_witness do |w|
+      with_dummy_top_transaction do |parent_tx|
+        top = parent_tx
+        parent_tx.transaction do |tx|
+          if tx.is_a?(DummyDriver::DummySavePointTransaction)
+            t = tx
+            w.check
+          end
+          raise DB::Rollback.new
+        end
+      end
+    end
+
+    t.rolledback.should be_true
+    t.committed.should be_false
+
+    top.rolledback.should be_false
+    top.committed.should be_true
+  end
+
+  it "only nested transaction with block from parent nested transaction should be rolledback if raise DB::Rollback" do
+    top = uninitialized DummyDriver::DummySavePointTransaction
+    t = uninitialized DummyDriver::DummySavePointTransaction
+
+    with_witness do |w|
+      with_dummy_nested_transaction do |parent_tx|
+        top = parent_tx
+        parent_tx.transaction do |tx|
+          if tx.is_a?(DummyDriver::DummySavePointTransaction)
+            t = tx
+            w.check
+          end
+          raise DB::Rollback.new
+        end
+      end
+    end
+
+    t.rolledback.should be_true
+    t.committed.should be_false
+
+    top.rolledback.should be_false
+    top.committed.should be_true
+  end
+
+  it "releasing result_set from within inner transaction should not return connection to pool" do
+    cnn = uninitialized DB::Connection
+    with_dummy do |db|
+      db.transaction do |tx|
+        tx.transaction do |inner|
+          cnn = inner.connection
+          cnn.scalar "1"
+          db.pool.is_available?(cnn).should be_false
+        end
+        db.pool.is_available?(cnn).should be_false
+      end
+      db.pool.is_available?(cnn).should be_true
+    end
+  end
+
+  it "releasing result_set from within inner inner transaction should not return connection to pool" do
+    cnn = uninitialized DB::Connection
+    with_dummy do |db|
+      db.transaction do |tx|
+        tx.transaction do |inner|
+          inner.transaction do |inner_inner|
+            cnn = inner_inner.connection
+            cnn.scalar "1"
+            db.pool.is_available?(cnn).should be_false
+          end
+          db.pool.is_available?(cnn).should be_false
+        end
+        db.pool.is_available?(cnn).should be_false
+      end
+      db.pool.is_available?(cnn).should be_true
+    end
+  end
+end

spec/transaction_spec.cr

@@ -0,0 +1,178 @@
+require "./spec_helper"
+
+private class FooException < Exception
+end
+
+describe DB::Transaction do
+  it "begin/commit transaction from connection" do
+    with_dummy_connection do |cnn|
+      tx = cnn.begin_transaction
+      tx.commit
+    end
+  end
+
+  it "begin/rollback transaction from connection" do
+    with_dummy_connection do |cnn|
+      tx = cnn.begin_transaction
+      tx.rollback
+    end
+  end
+
+  it "raise if begin over existing transaction" do
+    with_dummy_connection do |cnn|
+      cnn.begin_transaction
+      expect_raises(DB::Error, "There is an existing transaction in this connection") do
+        cnn.begin_transaction
+      end
+    end
+  end
+
+  it "allow sequential transactions" do
+    with_dummy_connection do |cnn|
+      tx = cnn.begin_transaction
+      tx.rollback
+
+      tx = cnn.begin_transaction
+      tx.commit
+    end
+  end
+
+  it "transaction with block from connection should be committed" do
+    t = uninitialized DummyDriver::DummyTransaction
+
+    with_witness do |w|
+      with_dummy_connection do |cnn|
+        cnn.transaction do |tx|
+          if tx.is_a?(DummyDriver::DummyTransaction)
+            t = tx
+            w.check
+          end
+        end
+      end
+    end
+
+    t.committed.should be_true
+    t.rolledback.should be_false
+  end
+
+  it "transaction with block from connection should be rolledback if raise DB::Rollback" do
+    t = uninitialized DummyDriver::DummyTransaction
+
+    with_witness do |w|
+      with_dummy_connection do |cnn|
+        cnn.transaction do |tx|
+          if tx.is_a?(DummyDriver::DummyTransaction)
+            t = tx
+            w.check
+          end
+          raise DB::Rollback.new
+        end
+      end
+    end
+
+    t.rolledback.should be_true
+    t.committed.should be_false
+  end
+
+  it "transaction with block from connection should be rolledback if raise" do
+    t = uninitialized DummyDriver::DummyTransaction
+
+    with_witness do |w|
+      with_dummy_connection do |cnn|
+        expect_raises(FooException) do
+          cnn.transaction do |tx|
+            if tx.is_a?(DummyDriver::DummyTransaction)
+              t = tx
+              w.check
+            end
+            raise FooException.new
+          end
+        end
+      end
+    end
+
+    t.rolledback.should be_true
+    t.committed.should be_false
+  end
+
+  it "transaction can be committed within block" do
+    with_dummy_connection do |cnn|
+      cnn.transaction do |tx|
+        tx.commit
+      end
+    end
+  end
+
+  it "transaction can be rolledback within block" do
+    with_dummy_connection do |cnn|
+      cnn.transaction do |tx|
+        tx.rollback
+      end
+    end
+  end
+
+  it "transaction can be rolledback within block and later raise" do
+    with_dummy_connection do |cnn|
+      expect_raises(FooException) do
+        cnn.transaction do |tx|
+          tx.rollback
+          raise FooException.new
+        end
+      end
+    end
+  end
+
+  it "transaction can be rolledback within block and later raise DB::Rollback without forwarding it" do
+    with_dummy_connection do |cnn|
+      cnn.transaction do |tx|
+        tx.rollback
+        raise DB::Rollback.new
+      end
+    end
+  end
+
+  it "transaction can't be committed twice" do
+    with_dummy_connection do |cnn|
+      cnn.transaction do |tx|
+        tx.commit
+        expect_raises(DB::Error, "Transaction already closed") do
+          tx.commit
+        end
+      end
+    end
+  end
+
+  it "transaction can't be rolledback twice" do
+    with_dummy_connection do |cnn|
+      cnn.transaction do |tx|
+        tx.rollback
+        expect_raises(DB::Error, "Transaction already closed") do
+          tx.rollback
+        end
+      end
+    end
+  end
+
+  it "return connection to pool after transaction block in db" do
+    DummyDriver::DummyConnection.clear_connections
+
+    with_dummy do |db|
+      db.transaction do |tx|
+        db.pool.is_available?(DummyDriver::DummyConnection.connections.first).should be_false
+      end
+      db.pool.is_available?(DummyDriver::DummyConnection.connections.first).should be_true
+    end
+  end
+
+  it "releasing result_set from within transaction should not return connection to pool" do
+    cnn = uninitialized DB::Connection
+    with_dummy do |db|
+      db.transaction do |tx|
+        cnn = tx.connection
+        cnn.scalar "1"
+        db.pool.is_available?(cnn).should be_false
+      end
+      db.pool.is_available?(cnn).should be_true
+    end
+  end
+end

src/db.cr

@@ -143,7 +143,10 @@ require "./db/session_methods"
 require "./db/disposable"
 require "./db/driver"
 require "./db/statement"
+require "./db/begin_transaction"
 require "./db/connection"
+require "./db/transaction"
+require "./db/statement"
 require "./db/pool_statement"
 require "./db/database"
 require "./db/pool_prepared_statement"

src/db/begin_transaction.cr

@@ -0,0 +1,26 @@
+module DB
+  module BeginTransaction
+    # Creates a transaction from the current context.
+    # If is expected that either `Transaction#commit` or `Transaction#rollback`
+    # are called explictly to release the context.
+    abstract def begin_transaction : Transaction
+
+    # yields a transaction from the current context.
+    # Query the database through `Transaction#connection` object.
+    # If an exception is thrown within the block a rollback is performed.
+    # The exception thrown is blubbled unless it is a `DB::Rollback`.
+    # From the yielded object `Transaction#commit` or `Transaction#rollback`
+    # can be called explicitly.
+    def transaction
+      tx = begin_transaction
+      begin
+        yield tx
+      rescue e
+        tx.rollback unless tx.closed?
+        raise e unless e.is_a?(DB::Rollback)
+      else
+        tx.commit unless tx.closed?
+      end
+    end
+  end
+end

src/db/connection.cr

@@ -21,10 +21,12 @@ module DB
   abstract class Connection
     include Disposable
     include SessionMethods(Connection, Statement)
+    include BeginTransaction
 
     # :nodoc:
     getter database
     @statements_cache = StringKeyCache(Statement).new
+    @transaction = false
     getter? prepared_statements : Bool
 
     def initialize(@database : Database)
@@ -42,10 +44,60 @@ module DB
     # :nodoc:
     abstract def build_unprepared_statement(query) : Statement
 
+    def begin_transaction
+      raise DB::Error.new("There is an existing transaction in this connection") if @transaction
+      @transaction = true
+      create_transaction
+    end
+
+    protected def create_transaction : Transaction
+      TopLevelTransaction.new(self)
+    end
+
     protected def do_close
       @statements_cache.each_value &.close
       @statements_cache.clear
       @database.pool.delete self
     end
+
+    # :nodoc:
+    def release_from_statement
+      @database.return_to_pool(self) unless @transaction
+    end
+
+    # :nodoc:
+    def release_from_transaction
+      @transaction = false
+    end
+
+    # :nodoc:
+    def perform_begin_transaction
+      self.unprepared.exec "BEGIN"
+    end
+
+    # :nodoc:
+    def perform_commit_transaction
+      self.unprepared.exec "COMMIT"
+    end
+
+    # :nodoc:
+    def perform_rollback_transaction
+      self.unprepared.exec "ROLLBACK"
+    end
+
+    # :nodoc:
+    def perform_create_savepoint(name)
+      self.unprepared.exec "SAVEPOINT #{name}"
+    end
+
+    # :nodoc:
+    def perform_release_savepoint(name)
+      self.unprepared.exec "RELEASE SAVEPOINT #{name}"
+    end
+
+    # :nodoc:
+    def perform_rollback_savepoint(name)
+      self.unprepared.exec "ROLLBACK TO #{name}"
+    end
   end
 end

src/db/database.cr

@@ -105,6 +105,16 @@ module DB
       end
     end
 
+    # yields a `Transaction` from a connection of the pool
+    # Refer to `BeginTransaction#transaction` for documentation.
+    def transaction
+      using_connection do |cnn|
+        cnn.transaction do |tx|
+          yield tx
+        end
+      end
+    end
+
     # :nodoc:
     def retry
       @pool.retry do

src/db/error.cr

@@ -17,4 +17,7 @@ module DB
     def initialize(@connection)
     end
   end
+
+  class Rollback < Exception
+  end
 end

src/db/statement.cr

@@ -59,7 +59,7 @@ module DB
     end
 
     def release_connection
-      @connection.database.return_to_pool(@connection)
+      @connection.release_from_statement
     end
 
     # See `QueryMethods#exec`

src/db/transaction.cr

@@ -0,0 +1,131 @@
+module DB
+  # Transactions should be started from `DB#transaction`, `Connection#transaction`
+  # or `Connection#begin_transaction`.
+  #
+  # Use `Transaction#connection` to submit statements to the database.
+  #
+  # Use `Transaction#commit` or `Transaction#rollback` to close the ongoing transaction
+  # explicitly. Or refer to `BeginTransaction#transaction` for documentation on how to
+  # use `#transaction(&block)` methods in `DB` and `Connection`.
+  #
+  # Nested transactions are supported by using sql `SAVEPOINT`. To start a nested
+  # transaction use `Transaction#transaction` or `Transaction#begin_transaction`.
+  #
+  abstract class Transaction
+    include Disposable
+    include BeginTransaction
+
+    abstract def connection : Connection
+
+    # commits the current transaction
+    def commit
+      close!
+    end
+
+    # rollbacks the current transaction
+    def rollback
+      close!
+    end
+
+    private def close!
+      raise DB::Error.new("Transaction already closed") if closed?
+      close
+    end
+
+    abstract def release_from_nested_transaction
+  end
+
+  class TopLevelTransaction < Transaction
+    getter connection
+    # :nodoc:
+    property savepoint_name : String? = nil
+
+    def initialize(@connection : Connection)
+      @nested_transaction = false
+      @connection.perform_begin_transaction
+    end
+
+    def commit
+      @connection.perform_commit_transaction
+      super
+    end
+
+    def rollback
+      @connection.perform_rollback_transaction
+      super
+    end
+
+    protected def do_close
+      connection.release_from_transaction
+    end
+
+    def begin_transaction : Transaction
+      raise DB::Error.new("There is an existing nested transaction in this transaction") if @nested_transaction
+      @nested_transaction = true
+      create_save_point_transaction(self)
+    end
+
+    # :nodoc:
+    def create_save_point_transaction(parent : Transaction) : SavePointTransaction
+      # TODO should we wrap this in a mutex?
+      previous_savepoint = @savepoint_name
+      savepoint_name = if previous_savepoint
+                         previous_savepoint.succ
+                       else
+                         # random prefix to avoid determinism
+                         "cr_#{@connection.object_id}_#{Random.rand(10_000)}_00001"
+                       end
+
+      @savepoint_name = savepoint_name
+
+      create_save_point_transaction(parent, savepoint_name)
+    end
+
+    protected def create_save_point_transaction(parent : Transaction, savepoint_name : String) : SavePointTransaction
+      SavePointTransaction.new(parent, savepoint_name)
+    end
+
+    # :nodoc:
+    def release_from_nested_transaction
+      @nested_transaction = false
+    end
+  end
+
+  class SavePointTransaction < Transaction
+    getter connection : Connection
+
+    def initialize(@parent : Transaction, @savepoint_name : String)
+      @nested_transaction = false
+      @connection = @parent.connection
+      @connection.perform_create_savepoint(@savepoint_name)
+    end
+
+    def commit
+      @connection.perform_release_savepoint(@savepoint_name)
+      super
+    end
+
+    def rollback
+      @connection.perform_rollback_savepoint(@savepoint_name)
+      super
+    end
+
+    protected def do_close
+      @parent.release_from_nested_transaction
+    end
+
+    def begin_transaction : Transaction
+      raise DB::Error.new("There is an existing nested transaction in this transaction") if @nested_transaction
+      @nested_transaction = true
+      create_save_point_transaction(self)
+    end
+
+    def create_save_point_transaction(parent : Transaction)
+      @parent.create_save_point_transaction(parent)
+    end
+
+    def release_from_nested_transaction
+      @nested_transaction = false
+    end
+  end
+end