Merge lp:~salgado/storm/fix-timeout-tracer into lp:storm

[Guilherme Salgado]

Add connection commit/rollback tracer hooks and use them in the TimeoutTracer
to reset connection._timeout_tracer_remaining_time.

We currently don't reset connection._timeout_tracer_remaining_time ever, so
when the connection is reused in a new transaction the TimeoutTracer will not
call set_statement_timeout if connection._timeout_tracer_remaining_time is
greater than what's returned by self.get_remaining_time().

This is specially likely to happen if it's the first query you run in the
transaction that times out, as in that case when the transaction rolls back
connection._timeout_tracer_remaining_time will hold a value very close to your
total transaction timeout.

Oh, and I've also removed some duplicated code (a test method and StubConnection).

[lp:~salgado/storm/fix-timeout-tracer]

451. By Guilherme Salgado on 2012-06-27

Remove some commented out code I left by accident

[Christopher Armstrong (radix)]

[1] I would prefer for the new tests in test_tracer (test_connection_commit_hook and test_connection_rollback_hook) to be blackbox instead of whitebox, since it doesn't look like that should be too hard (or at least *less* whiteboxy; something that actually invokes some connection logic and observes that the bug that you observed does not occur). Even if you don't change it, it looks like storm has a convention of using "test_wb_" for whitebox tests which you should use here.

[2] There are no tests for the changes to DebugTracer (and there is already a suite of tests for it, so it should be easy to add).

Looks good!

review: Approve

[Gavin Panella (allenap)]

Nothing to add to what Christopher has already said. Nice :)

review: Approve

[lp:~salgado/storm/fix-timeout-tracer]

452. By Guilherme Salgado on 2012-06-28

A couple tests to the new DebugTracer methods

453. By Guilherme Salgado on 2012-06-28

Rewrite test_connection_commit_hook and test_connection_rollback_hook to be black box; we now use a real DB connection and execute/commit/rollback, making sure the statement timeout is set when it should

[Guilherme Salgado (salgado)]

On 27/06/12 21:31, Christopher Armstrong wrote:
> Review: Approve
>
> [1] I would prefer for the new tests in test_tracer (test_connection_commit_hook and test_connection_rollback_hook) to be blackbox instead of whitebox, since it doesn't look like that should be too hard (or at least *less* whiteboxy; something that actually invokes some connection logic and observes that the bug that you observed does not occur). Even if you don't change it, it looks like storm has a convention of using "test_wb_" for whitebox tests which you should use here.

Ok, I rewrote them. They're now in a separate TestCase since I'm calling
conn.execute()/etc against a real DB, so I'm not sure tests/tracer.py is
the right home for that. Also, I think I could simplify its
setUp/tearDown a bit by inheriting from DatabaseTest, if you think it's
worth it.

> [2] There are no tests for the changes to DebugTracer (and there is already a suite of tests for it, so it should be easy to add).

I completely missed that; just added a couple more tests there.

Thanks for the review!

[Christopher Armstrong (radix)]

I think the branch looks great now, +1.

=== modified file 'storm/database.py'
--- storm/database.py	2012-03-02 20:36:48 +0000
+++ storm/database.py	2012-06-28 14:23:21 +0000
@@ -281,15 +281,18 @@
             Reconnection happens automatically on rollback.
 
         """
-        self._ensure_connected()
-        if xid:
-            raw_xid = self._raw_xid(xid)
-            self._check_disconnect(self._raw_connection.tpc_commit, raw_xid)
-        elif self._two_phase_transaction:
-            self._check_disconnect(self._raw_connection.tpc_commit)
-            self._two_phase_transaction = False
-        else:
-            self._check_disconnect(self._raw_connection.commit)
+        try:
+            self._ensure_connected()
+            if xid:
+                raw_xid = self._raw_xid(xid)
+                self._check_disconnect(self._raw_connection.tpc_commit, raw_xid)
+            elif self._two_phase_transaction:
+                self._check_disconnect(self._raw_connection.tpc_commit)
+                self._two_phase_transaction = False
+            else:
+                self._check_disconnect(self._raw_connection.commit)
+        finally:
+            self._check_disconnect(trace, "connection_commit", self, xid)
 
     def recover(self):
         """Return a list of L{Xid}s representing pending transactions."""
@@ -305,27 +308,30 @@
              transaction to rollback. This form should be called outside
              of a transaction, and is intended for use in recovery.
         """
-        if self._state == STATE_CONNECTED:
-            try:
-                if xid:
-                    raw_xid = self._raw_xid(xid)
-                    self._raw_connection.tpc_rollback(raw_xid)
-                elif self._two_phase_transaction:
-                    self._raw_connection.tpc_rollback()
+        try:
+            if self._state == STATE_CONNECTED:
+                try:
+                    if xid:
+                        raw_xid = self._raw_xid(xid)
+                        self._raw_connection.tpc_rollback(raw_xid)
+                    elif self._two_phase_transaction:
+                        self._raw_connection.tpc_rollback()
+                    else:
+                        self._raw_connection.rollback()
+                except Error, exc:
+                    if self.is_disconnection_error(exc):
+                        self._raw_connection = None
+                        self._state = STATE_RECONNECT
+                        self._two_phase_transaction = False
+                    else:
+                        raise
                 else:
-                    self._raw_connection.rollback()
-            except Error, exc:
-                if self.is_disconnection_error(exc):
-                    self._raw_connection = None
-                    self._state = STATE_RECONNECT
                     self._two_phase_transaction = False
-                else:
-                    raise
             else:
                 self._two_phase_transaction = False
-        else:
-            self._two_phase_transaction = False
-            self._state = STATE_RECONNECT
+                self._state = STATE_RECONNECT
+        finally:
+            self._check_disconnect(trace, "connection_rollback", self, xid)
 
     @staticmethod
     def to_database(params):

=== modified file 'storm/tracer.py'
--- storm/tracer.py	2012-03-21 14:43:26 +0000
+++ storm/tracer.py	2012-06-28 14:23:21 +0000
@@ -42,6 +42,16 @@
         self._stream.write("[%s] DONE\n" % time)
         self._stream.flush()
 
+    def connection_commit(self, connection, xid=None):
+        time = datetime.now().isoformat()[11:]
+        self._stream.write("[%s] COMMIT xid=%s\n" % (time, xid))
+        self._stream.flush()
+
+    def connection_rollback(self, connection, xid=None):
+        time = datetime.now().isoformat()[11:]
+        self._stream.write("[%s] ROLLBACK xid=%s\n" % (time, xid))
+        self._stream.flush()
+
 
 class TimeoutTracer(object):
     """Provide a timeout facility for connections to prevent rogue operations.
@@ -87,6 +97,28 @@
         raise NotImplementedError("%s.connection_raw_execute_error() must be "
                                   "implemented" % self.__class__.__name__)
 
+    def connection_commit(self, connection, xid=None):
+        """Reset Connection._timeout_tracer_remaining_time.
+
+        @param connection: The L{Connection} to the database.
+        @param xid: Optionally the L{Xid} of a previously prepared
+             transaction to commit.
+        """
+        self._reset_timeout_tracer_remaining_time(connection)
+
+    def connection_rollback(self, connection, xid=None):
+        """Reset Connection._timeout_tracer_remaining_time.
+
+        @param connection: The L{Connection} to the database.
+        @param xid: Optionally the L{Xid} of a previously prepared
+             transaction to rollback.
+        """
+        self._reset_timeout_tracer_remaining_time(connection)
+
+    def _reset_timeout_tracer_remaining_time(self, connection):
+        """Set connection._timeout_tracer_remaining_time to 0."""
+        connection._timeout_tracer_remaining_time = 0
+
     def set_statement_timeout(self, raw_cursor, remaining_time):
         """Perform the timeout setup in the raw cursor.
 

=== modified file 'tests/database.py'
--- tests/database.py	2011-09-22 09:16:49 +0000
+++ tests/database.py	2012-06-28 14:23:21 +0000
@@ -118,6 +118,12 @@
         self.seen.append(("ERROR", connection, type(raw_cursor),
                           statement, params, error))
 
+    def connection_commit(self, connection, xid=None):
+        self.seen.append(("COMMIT", connection, xid))
+
+    def connection_rollback(self, connection, xid=None):
+        self.seen.append(("ROLLBACK", connection, xid))
+
 
 class DatabaseTest(TestHelper):
 
@@ -159,10 +165,6 @@
         self.assertEquals(self.executed, [("something", marker), "RCLOSE"])
 
     def test_execute_convert_param_style(self):
-        self.connection.execute("'?' ? '?' ? '?'")
-        self.assertEquals(self.executed, [("'?' ? '?' ? '?'", marker)])
-
-    def test_execute_convert_param_style(self):
         class MyConnection(Connection):
             param_mark = "%s"
         connection = MyConnection(self.database)
@@ -282,10 +284,24 @@
         self.connection.commit()
         self.assertEquals(self.executed, ["COMMIT"])
 
+    def test_commit_tracing(self):
+        self.assertMethodsMatch(FakeTracer, DebugTracer)
+        tracer = FakeTracer()
+        install_tracer(tracer)
+        self.connection.commit()
+        self.assertEquals(tracer.seen, [("COMMIT", self.connection, None)])
+
     def test_rollback(self):
         self.connection.rollback()
         self.assertEquals(self.executed, ["ROLLBACK"])
 
+    def test_rollback_tracing(self):
+        self.assertMethodsMatch(FakeTracer, DebugTracer)
+        tracer = FakeTracer()
+        install_tracer(tracer)
+        self.connection.rollback()
+        self.assertEquals(tracer.seen, [("ROLLBACK", self.connection, None)])
+
     def test_close(self):
         self.connection.close()
         self.assertEquals(self.executed, ["CCLOSE"])

=== modified file 'tests/tracer.py'
--- tests/tracer.py	2012-03-21 15:59:24 +0000
+++ tests/tracer.py	2012-06-28 14:23:21 +0000
@@ -1,4 +1,5 @@
 import datetime
+import os
 import sys
 from unittest import TestCase
 
@@ -23,7 +24,7 @@
                           remove_tracer_type, remove_all_tracers, debug,
                           BaseStatementTracer, DebugTracer, TimeoutTracer,
                           TimelineTracer, TimeoutError, _tracers)
-from storm.database import Connection
+from storm.database import Connection, create_database
 from storm.expr import Variable
 
 from tests.helper import TestHelper
@@ -209,6 +210,24 @@
         self.tracer.connection_raw_execute_success(connection, raw_cursor,
                                                    statement, params)
 
+    def test_connection_commit(self):
+        self.stream.write("[04:05:06.000007] COMMIT xid=None\n")
+        self.stream.flush()
+        self.mocker.replay()
+
+        connection = "CONNECTION"
+
+        self.tracer.connection_commit(connection)
+
+    def test_connection_rollback(self):
+        self.stream.write("[04:05:06.000007] ROLLBACK xid=None\n")
+        self.stream.flush()
+        self.mocker.replay()
+
+        connection = "CONNECTION"
+
+        self.tracer.connection_rollback(connection)
+
 
 class TimeoutTracerTestBase(TestHelper):
 
@@ -363,12 +382,73 @@
         self.execute()
 
 
+class TimeoutTracerWithDBTest(TestHelper):
+
+    def setUp(self):
+        super(TimeoutTracerWithDBTest, self).setUp()
+        self.tracer = StuckInTimeTimeoutTracer(10)
+        install_tracer(self.tracer)
+        database = create_database(os.environ["STORM_POSTGRES_URI"])
+        self.connection = database.connect()
+
+    def tearDown(self):
+        super(TimeoutTracerWithDBTest, self).tearDown()
+        remove_tracer(self.tracer)
+        self.connection.close()
+
+    def is_supported(self):
+        return bool(os.environ.get("STORM_POSTGRES_URI"))
+
+    def test_timeout_set_on_beginning_of_new_transaction__commit(self):
+        """Check that we set the statement timeout before the first query of a
+        transaction regardless of the remaining time left by previous
+        transactions.
+
+        When we reuse a connection for a different transaction, the remaining
+        time of a previous transaction (which is stored in the connection)
+        could cause the first query in that transaction to run with no
+        timeout. This test makes sure that doesn't happen.
+        """
+        self.connection.execute('SELECT 1')
+        self.assertEqual([10], self.tracer.set_statement_timeout_calls)
+
+        self.connection.commit()
+
+        self.connection.execute('SELECT 1')
+        self.assertEqual([10, 10], self.tracer.set_statement_timeout_calls)
+
+    def test_timeout_set_on_beginning_of_new_transaction__rollback(self):
+        """Same as the test above, but here we rollback the first tx."""
+        self.connection.execute('SELECT 1')
+        self.assertEqual([10], self.tracer.set_statement_timeout_calls)
+
+        self.connection.rollback()
+
+        self.connection.execute('SELECT 1')
+        self.assertEqual([10, 10], self.tracer.set_statement_timeout_calls)
+
+
+class StuckInTimeTimeoutTracer(TimeoutTracer):
+
+    def __init__(self, fixed_remaining_time):
+        super(StuckInTimeTimeoutTracer, self).__init__()
+        self.set_statement_timeout_calls = []
+        self.fixed_remaining_time = fixed_remaining_time
+
+    def get_remaining_time(self):
+        return self.fixed_remaining_time
+
+    def set_statement_timeout(self, raw_cursor, remaining_time):
+        self.set_statement_timeout_calls.append(remaining_time)
+
+
 class StubConnection(Connection):
 
     def __init__(self):
         self._database = None
         self._event = None
         self._raw_connection = None
+        self.name = 'Foo'
 
 
 class BaseStatementTracerTest(TestCase):
@@ -502,13 +582,6 @@
         self.assertEqual('statement', self.timeline.actions[-1].detail)
 
     def test_category_from_prefix_and_connection_name(self):
-        class StubConnection(Connection):
-
-            def __init__(self):
-                self._database = None
-                self._event = None
-                self._raw_connection = None
-                self.name = 'Foo'
         tracer = TimelineTracer(self.factory, prefix='bar-')
         tracer._expanded_raw_execute(StubConnection(), 'cursor', 'statement')
         self.assertEqual('bar-Foo', self.timeline.actions[-1].category)