Merge lp:~allenap/maas/retry-integrity-errors into lp:~maas-committers/maas/trunk

[Gavin Panella]

Retry transactions when they fail with unique violation errors.

This also makes explicit retrying via request_transaction_retry a separate thing rather than a piggyback on serialization failures.

[Blake Rouse (blake-rouse)]

Looks really good. The test case for creating the exception is awesome. Just some comments and 1 question.

review: Approve

[Gavin Panella (allenap)]

Thanks!

[Gavin Panella (allenap)]

Because of the potential for fallout at this late time in the release cycle I'm running this through CI before landing.

[Gavin Panella (allenap)]

All good in CI.

=== modified file 'src/maasserver/testing/testcase.py'
--- src/maasserver/testing/testcase.py	2016-05-24 12:51:54 +0000
+++ src/maasserver/testing/testcase.py	2016-06-17 14:52:22 +0000
@@ -9,9 +9,10 @@
     'SeleniumTestCase',
     'SerializationFailureTestCase',
     'TestWithoutCrochetMixin',
+    'UniqueViolationTestCase',
     ]
 
-from contextlib import closing
+from itertools import count
 import socketserver
 import sys
 import threading
@@ -28,13 +29,19 @@
     connection,
     transaction,
 )
-from django.db.utils import OperationalError
+from django.db.utils import (
+    IntegrityError,
+    OperationalError,
+)
 from fixtures import Fixture
 from maasserver.fields import register_mac_type
 from maasserver.testing.factory import factory
 from maasserver.testing.orm import PostCommitHooksTestMixin
 from maasserver.testing.testclient import MAASSensibleClient
-from maasserver.utils.orm import is_serialization_failure
+from maasserver.utils.orm import (
+    is_serialization_failure,
+    is_unique_violation,
+)
 from maastesting.djangotestcase import (
     DjangoTestCase,
     DjangoTransactionTestCase,
@@ -234,11 +241,11 @@
         DjangoTransactionTestCase, PostCommitHooksTestMixin):
 
     def create_stest_table(self):
-        with closing(connection.cursor()) as cursor:
+        with connection.cursor() as cursor:
             cursor.execute("CREATE TABLE IF NOT EXISTS stest (a INTEGER)")
 
     def drop_stest_table(self):
-        with closing(connection.cursor()) as cursor:
+        with connection.cursor() as cursor:
             cursor.execute("DROP TABLE IF EXISTS stest")
 
     def setUp(self):
@@ -247,7 +254,7 @@
         # Put something into the stest table upon which to trigger a
         # serialization failure.
         with transaction.atomic():
-            with closing(connection.cursor()) as cursor:
+            with connection.cursor() as cursor:
                 cursor.execute("INSERT INTO stest VALUES (1)")
 
     def tearDown(self):
@@ -258,7 +265,7 @@
         """Trigger an honest, from the database, serialization failure."""
         # Helper to switch the transaction to SERIALIZABLE.
         def set_serializable():
-            with closing(connection.cursor()) as cursor:
+            with connection.cursor() as cursor:
                 cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
 
         # Perform a conflicting update. This must run in a separate thread. It
@@ -269,7 +276,7 @@
         def do_conflicting_update():
             try:
                 with transaction.atomic():
-                    with closing(connection.cursor()) as cursor:
+                    with connection.cursor() as cursor:
                         cursor.execute("UPDATE stest SET a = 2")
             finally:
                 close_old_connections()
@@ -278,7 +285,7 @@
             # Fetch something first. This ensures that we're inside the
             # transaction, and that the database has a reference point for
             # calculating serialization failures.
-            with closing(connection.cursor()) as cursor:
+            with connection.cursor() as cursor:
                 cursor.execute("SELECT * FROM stest")
                 cursor.fetchall()
 
@@ -290,7 +297,7 @@
             # Updating the same rows as do_conflicting_update() did will
             # trigger a serialization failure. We have to check the __cause__
             # to confirm the failure type as reported by PostgreSQL.
-            with closing(connection.cursor()) as cursor:
+            with connection.cursor() as cursor:
                 cursor.execute("UPDATE stest SET a = 4")
 
         if connection.in_atomic_block:
@@ -312,3 +319,148 @@
                 return sys.exc_info()
             else:
                 raise
+
+
+class UniqueViolationTestCase(
+        DjangoTransactionTestCase, PostCommitHooksTestMixin):
+
+    def create_uvtest_table(self):
+        with connection.cursor() as cursor:
+            cursor.execute("DROP TABLE IF EXISTS uvtest")
+            cursor.execute("CREATE TABLE uvtest (a INTEGER PRIMARY KEY)")
+
+    def drop_uvtest_table(self):
+        with connection.cursor() as cursor:
+            cursor.execute("DROP TABLE IF EXISTS uvtest")
+
+    def setUp(self):
+        super(UniqueViolationTestCase, self).setUp()
+        self.conflicting_values = count(1)
+        self.create_uvtest_table()
+
+    def tearDown(self):
+        super(UniqueViolationTestCase, self).tearDown()
+        self.drop_uvtest_table()
+
+    def cause_unique_violation(self):
+        """Trigger an honest, from the database, unique violation.
+
+        This may appear needlessly elaborate, but it's for a good reason.
+        Indexes in PostgreSQL are a bit weird; they don't fully support MVCC
+        so it's possible for situations like the following:
+
+          CREATE TABLE foo (id SERIAL PRIMARY KEY);
+          -- Session A:
+          BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+          INSERT INTO foo (id) VALUES (1);
+          -- Session B:
+          BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+          SELECT id FROM foo;  -- Nothing.
+          INSERT INTO foo (id) VALUES (1);  -- Hangs.
+          -- Session A:
+          COMMIT;
+          -- Session B:
+          ERROR:  duplicate key value violates unique constraint "..."
+          DETAIL:  Key (id)=(1) already exists.
+
+        Two things to note:
+
+          1. Session B hangs when there's a potential conflict on id's index.
+
+          2. Session B fails with a duplicate key error.
+
+        Both differ from expectations:
+
+          1. I would expect the transaction to continue optimistically and
+             only fail if session A commits.
+
+          2. I would expect a serialisation failure instead.
+
+        This method jumps through hoops to reproduce the situation above so
+        that we're testing against PostgreSQL's exact behaviour as of today,
+        not the behaviour that we observed at a single moment in time.
+        PostgreSQL may change its behaviour in later versions and this test
+        ought to tell us about it.
+
+        """
+        # Helper to switch the transaction to REPEATABLE READ.
+        def set_repeatable_read():
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    "SET TRANSACTION ISOLATION LEVEL "
+                    "REPEATABLE READ")
+
+        # Both threads / database sessions will attempt to insert this.
+        conflicting_value = next(self.conflicting_values)
+
+        # Perform a conflicting insert. This must run in a separate thread. It
+        # also must begin after the beginning of the transaction in which we
+        # will trigger a unique violation AND commit before that other
+        # transaction commits. This doesn't need to run with any special
+        # isolation; it just needs to be in a transaction.
+        def do_conflicting_insert():
+            try:
+                with transaction.atomic():
+                    with connection.cursor() as cursor:
+                        cursor.execute(
+                            "INSERT INTO uvtest VALUES (%s)",
+                            [conflicting_value])
+            finally:
+                close_old_connections()
+
+        def trigger_unique_violation():
+            # Fetch something first. This ensures that we're inside the
+            # transaction, and so the database has a reference point for
+            # repeatable reads.
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    "SELECT 1 FROM uvtest WHERE a = %s",
+                    [conflicting_value])
+                self.assertIsNone(cursor.fetchone(), (
+                    "We've seen through PostgreSQL impenetrable transaction "
+                    "isolation — or so we once thought — to witness a "
+                    "conflicting value from another database session. "
+                    "Needless to say, this requires investigation."))
+
+            # Run do_conflicting_insert() in a separate thread and wait for it
+            # to commit and return.
+            thread = threading.Thread(target=do_conflicting_insert)
+            thread.start()
+            thread.join()
+
+            # Still no sign of that conflicting value from here.
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    "SELECT 1 FROM uvtest WHERE a = %s",
+                    [conflicting_value])
+                self.assertIsNone(cursor.fetchone(), (
+                    "PostgreSQL, once thought of highly in transactional "
+                    "circles, has dropped its kimono and disgraced itself "
+                    "with its wanton exhibition of conflicting values from "
+                    "another's session."))
+
+            # Inserting the same row will trigger a unique violation.
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    "INSERT INTO uvtest VALUES (%s)",
+                    [conflicting_value])
+
+        if connection.in_atomic_block:
+            # We're already in a transaction.
+            set_repeatable_read()
+            trigger_unique_violation()
+        else:
+            # Start a transaction in this thread.
+            with transaction.atomic():
+                set_repeatable_read()
+                trigger_unique_violation()
+
+    def capture_unique_violation(self):
+        """Trigger a unique violation, return its ``exc_info`` tuple."""
+        try:
+            self.cause_unique_violation()
+        except IntegrityError as e:
+            if is_unique_violation(e):
+                return sys.exc_info()
+            else:
+                raise

=== modified file 'src/maasserver/utils/orm.py'
--- src/maasserver/utils/orm.py	2016-03-28 13:54:47 +0000
+++ src/maasserver/utils/orm.py	2016-06-17 14:52:22 +0000
@@ -16,8 +16,10 @@
     'is_deadlock_failure',
     'is_retryable_failure',
     'is_serialization_failure',
+    'is_unique_violation',
     'make_deadlock_failure',
     'make_serialization_failure',
+    'make_unique_violation',
     'post_commit',
     'post_commit_do',
     'psql_array',
@@ -54,7 +56,11 @@
 )
 from django.db.models import Q
 from django.db.transaction import TransactionManagementError
-from django.db.utils import OperationalError
+from django.db.utils import (
+    DatabaseError,
+    IntegrityError,
+    OperationalError,
+)
 from django.http import Http404
 from maasserver.exceptions import (
     MAASAPIBadRequest,
@@ -72,6 +78,7 @@
 from psycopg2.errorcodes import (
     DEADLOCK_DETECTED,
     SERIALIZATION_FAILURE,
+    UNIQUE_VIOLATION,
 )
 from twisted.internet.defer import Deferred
 
@@ -273,6 +280,68 @@
     return get_psycopg2_deadlock_exception(exception) is not None
 
 
+def get_psycopg2_unique_violation_exception(exception):
+    """Return the root-cause if `exception` is a unique violation.
+
+    PostgreSQL sets a specific error code, "23505", when a transaction breaks
+    because of a unique violation.
+
+    :return: The underlying `psycopg2.Error` if it's a unique violation, or
+    `None` if there isn't one.
+    """
+    exception = get_psycopg2_exception(exception)
+    if exception is None:
+        return None
+    elif exception.pgcode == UNIQUE_VIOLATION:
+        return exception
+    else:
+        return None
+
+
+def is_unique_violation(exception):
+    """Does `exception` represent a unique violation?
+
+    PostgreSQL sets a specific error code, "23505", when a transaction breaks
+    because of a unique violation.
+    """
+    return get_psycopg2_unique_violation_exception(exception) is not None
+
+
+class UniqueViolation(psycopg2.IntegrityError):
+    """Explicit serialization failure.
+
+    A real unique violation, arising out of psycopg2 (and thus signalled from
+    the database) would *NOT* be an instance of this class. However, it is not
+    obvious how to create a `psycopg2.IntegrityError` with ``pgcode`` set to
+    `UNIQUE_VIOLATION` without subclassing. I suspect only the C interface can
+    do that.
+    """
+    pgcode = UNIQUE_VIOLATION
+
+
+def make_unique_violation():
+    """Make a serialization exception.
+
+    Artificially construct an exception that resembles what Django's ORM would
+    raise when PostgreSQL fails a transaction because of a unique violation.
+
+    :returns: an instance of :py:class:`IntegrityError` that will pass the
+        `is_unique_violation` predicate.
+    """
+    exception = IntegrityError()
+    exception.__cause__ = UniqueViolation()
+    assert is_unique_violation(exception)
+    return exception
+
+
+class RetryTransaction(BaseException):
+    """An explicit request that the transaction be retried."""
+
+
+class TooManyRetries(Exception):
+    """A transaction retry has been requested too many times."""
+
+
 def request_transaction_retry():
     """Raise a serialization exception.
 
@@ -280,15 +349,24 @@
     this, and then retrying the transaction, though it may choose to re-raise
     the error if too many retries have already been attempted.
 
-    :raises OperationalError:
+    :raise RetryTransaction:
     """
-    raise make_serialization_failure()
+    raise RetryTransaction()
 
 
 def is_retryable_failure(exception):
-    """Does `exception` represent a serialization or deadlock failure?"""
+    """Does `exception` represent a retryable failure?
+
+    This does NOT include requested retries, i.e. `RetryTransaction`.
+
+    :param exception: An instance of :class:`DatabaseError` or one of its
+        subclasses.
+    """
     return (
-        is_serialization_failure(exception) or is_deadlock_failure(exception))
+        is_serialization_failure(exception) or
+        is_deadlock_failure(exception) or
+        is_unique_violation(exception)
+    )
 
 
 def gen_retry_intervals(base=0.01, rate=2.5, maximum=10.0):
@@ -341,14 +419,22 @@
         for _ in range(9):
             try:
                 return func(*args, **kwargs)
-            except OperationalError as error:
+            except RetryTransaction:
+                reset()  # Which may do nothing.
+                sleep(next(intervals))
+            except DatabaseError as error:
                 if is_retryable_failure(error):
                     reset()  # Which may do nothing.
                     sleep(next(intervals))
                 else:
                     raise
         else:
-            return func(*args, **kwargs)
+            try:
+                return func(*args, **kwargs)
+            except RetryTransaction:
+                raise TooManyRetries(
+                    "This transaction has already been attempted "
+                    "multiple times; giving up.")
     return retrier
 
 

=== modified file 'src/maasserver/utils/tests/test_orm.py'
--- src/maasserver/utils/tests/test_orm.py	2016-05-12 19:07:37 +0000
+++ src/maasserver/utils/tests/test_orm.py	2016-06-17 14:52:22 +0000
@@ -26,12 +26,16 @@
 )
 from django.db.backends.base.base import BaseDatabaseWrapper
 from django.db.transaction import TransactionManagementError
-from django.db.utils import OperationalError
+from django.db.utils import (
+    IntegrityError,
+    OperationalError,
+)
 from maasserver.models import Node
 from maasserver.testing.testcase import (
     MAASServerTestCase,
     MAASTransactionServerTestCase,
     SerializationFailureTestCase,
+    UniqueViolationTestCase,
 )
 from maasserver.utils import orm
 from maasserver.utils.orm import (
@@ -46,11 +50,12 @@
     get_psycopg2_deadlock_exception,
     get_psycopg2_exception,
     get_psycopg2_serialization_exception,
+    get_psycopg2_unique_violation_exception,
     in_transaction,
     is_deadlock_failure,
     is_retryable_failure,
     is_serialization_failure,
-    make_serialization_failure,
+    is_unique_violation,
     post_commit,
     post_commit_do,
     post_commit_hooks,
@@ -81,6 +86,7 @@
 from psycopg2.errorcodes import (
     DEADLOCK_DETECTED,
     SERIALIZATION_FAILURE,
+    UNIQUE_VIOLATION,
 )
 from testtools import ExpectedException
 from testtools.matchers import (
@@ -221,6 +227,16 @@
             SERIALIZATION_FAILURE, error.__cause__.pgcode)
 
 
+class TestUniqueViolation(UniqueViolationTestCase):
+    """Detecting UNIQUE_VIOLATION failures."""
+
+    def test_unique_violation_detectable_via_error_cause(self):
+        error = self.assertRaises(
+            IntegrityError, self.cause_unique_violation)
+        self.assertEqual(
+            UNIQUE_VIOLATION, error.__cause__.pgcode)
+
+
 class TestGetPsycopg2Exception(MAASTestCase):
     """Tests for `get_psycopg2_exception`."""
 
@@ -281,6 +297,25 @@
             get_psycopg2_deadlock_exception(exception))
 
 
+class TestGetPsycopg2UniqueViolationException(MAASTestCase):
+    """Tests for `get_psycopg2_unique_violation_exception`."""
+
+    def test__returns_None_for_plain_psycopg2_error(self):
+        exception = psycopg2.Error()
+        self.assertIsNone(get_psycopg2_unique_violation_exception(exception))
+
+    def test__returns_None_for_other_error(self):
+        exception = factory.make_exception()
+        self.assertIsNone(get_psycopg2_unique_violation_exception(exception))
+
+    def test__returns_psycopg2_error_root_cause(self):
+        exception = Exception()
+        exception.__cause__ = orm.UniqueViolation()
+        self.assertIs(
+            exception.__cause__,
+            get_psycopg2_unique_violation_exception(exception))
+
+
 class TestIsSerializationFailure(SerializationFailureTestCase):
     """Tests relating to MAAS's use of SERIALIZABLE isolation."""
 
@@ -340,6 +375,36 @@
         self.assertFalse(is_deadlock_failure(error))
 
 
+class TestIsUniqueViolation(UniqueViolationTestCase):
+    """Tests relating to MAAS's identification of unique violations."""
+
+    def test_detects_integrity_error_with_matching_cause(self):
+        error = self.assertRaises(
+            IntegrityError, self.cause_unique_violation)
+        self.assertTrue(is_unique_violation(error))
+
+    def test_rejects_integrity_error_without_matching_cause(self):
+        error = IntegrityError()
+        cause = self.patch(error, "__cause__", Exception())
+        cause.pgcode = factory.make_name("pgcode")
+        self.assertFalse(is_unique_violation(error))
+
+    def test_rejects_integrity_error_with_unrelated_cause(self):
+        error = IntegrityError()
+        error.__cause__ = Exception()
+        self.assertFalse(is_unique_violation(error))
+
+    def test_rejects_integrity_error_without_cause(self):
+        error = IntegrityError()
+        self.assertFalse(is_unique_violation(error))
+
+    def test_rejects_non_integrity_error_with_matching_cause(self):
+        error = factory.make_exception()
+        cause = self.patch(error, "__cause__", Exception())
+        cause.pgcode = UNIQUE_VIOLATION
+        self.assertFalse(is_unique_violation(error))
+
+
 class TestIsRetryableFailure(MAASTestCase):
     """Tests relating to MAAS's use of catching retryable failures."""
 
@@ -351,33 +416,58 @@
         error = orm.make_deadlock_failure()
         self.assertTrue(is_retryable_failure(error))
 
+    def test_detects_unique_violation(self):
+        error = orm.make_unique_violation()
+        self.assertTrue(is_retryable_failure(error))
+
     def test_rejects_operational_error_without_matching_cause(self):
         error = OperationalError()
         cause = self.patch(error, "__cause__", Exception())
         cause.pgcode = factory.make_name("pgcode")
         self.assertFalse(is_retryable_failure(error))
 
+    def test_rejects_integrity_error_without_matching_cause(self):
+        error = IntegrityError()
+        cause = self.patch(error, "__cause__", Exception())
+        cause.pgcode = factory.make_name("pgcode")
+        self.assertFalse(is_retryable_failure(error))
+
     def test_rejects_operational_error_with_unrelated_cause(self):
         error = OperationalError()
         error.__cause__ = Exception()
         self.assertFalse(is_retryable_failure(error))
 
+    def test_rejects_integrity_error_with_unrelated_cause(self):
+        error = IntegrityError()
+        error.__cause__ = Exception()
+        self.assertFalse(is_retryable_failure(error))
+
     def test_rejects_operational_error_without_cause(self):
         error = OperationalError()
         self.assertFalse(is_retryable_failure(error))
 
-    def test_rejects_non_operational_error_with_cause_serialization(self):
+    def test_rejects_integrity_error_without_cause(self):
+        error = IntegrityError()
+        self.assertFalse(is_retryable_failure(error))
+
+    def test_rejects_non_database_error_with_cause_serialization(self):
         error = factory.make_exception()
         cause = self.patch(error, "__cause__", Exception())
         cause.pgcode = SERIALIZATION_FAILURE
         self.assertFalse(is_retryable_failure(error))
 
-    def test_rejects_non_operational_error_with_cause_deadlock(self):
+    def test_rejects_non_database_error_with_cause_deadlock(self):
         error = factory.make_exception()
         cause = self.patch(error, "__cause__", Exception())
         cause.pgcode = DEADLOCK_DETECTED
         self.assertFalse(is_retryable_failure(error))
 
+    def test_rejects_non_database_error_with_cause_unique_violation(self):
+        error = factory.make_exception()
+        cause = self.patch(error, "__cause__", Exception())
+        cause.pgcode = UNIQUE_VIOLATION
+        self.assertFalse(is_retryable_failure(error))
+
 
 class TestRetryOnRetryableFailure(SerializationFailureTestCase):
 
@@ -418,6 +508,36 @@
         self.assertEqual(sentinel.result, function_wrapped())
         self.assertThat(function, MockCallsMatch(call(), call()))
 
+    def test_retries_on_unique_violation(self):
+        function = self.make_mock_function()
+        function.side_effect = orm.make_unique_violation()
+        function_wrapped = retry_on_retryable_failure(function)
+        self.assertRaises(IntegrityError, function_wrapped)
+        expected_calls = [call()] * 10
+        self.assertThat(function, MockCallsMatch(*expected_calls))
+
+    def test_retries_on_unique_violation_until_successful(self):
+        function = self.make_mock_function()
+        function.side_effect = [orm.make_unique_violation(), sentinel.result]
+        function_wrapped = retry_on_retryable_failure(function)
+        self.assertEqual(sentinel.result, function_wrapped())
+        self.assertThat(function, MockCallsMatch(call(), call()))
+
+    def test_retries_on_retry_transaction(self):
+        function = self.make_mock_function()
+        function.side_effect = orm.RetryTransaction()
+        function_wrapped = retry_on_retryable_failure(function)
+        self.assertRaises(orm.TooManyRetries, function_wrapped)
+        expected_calls = [call()] * 10
+        self.assertThat(function, MockCallsMatch(*expected_calls))
+
+    def test_retries_on_retry_transaction_until_successful(self):
+        function = self.make_mock_function()
+        function.side_effect = [orm.RetryTransaction(), sentinel.result]
+        function_wrapped = retry_on_retryable_failure(function)
+        self.assertEqual(sentinel.result, function_wrapped())
+        self.assertThat(function, MockCallsMatch(call(), call()))
+
     def test_passes_args_to_wrapped_function(self):
         function = lambda a, b: (a, b)
         function_wrapped = retry_on_retryable_failure(function)
@@ -450,19 +570,34 @@
     """Tests for `make_serialization_failure`."""
 
     def test__makes_a_serialization_failure(self):
-        exception = make_serialization_failure()
+        exception = orm.make_serialization_failure()
         self.assertThat(exception, MatchesPredicate(
             is_serialization_failure, "%r is not a serialization failure."))
 
 
+class TestMakeDeadlockFailure(MAASTestCase):
+    """Tests for `make_deadlock_failure`."""
+
+    def test__makes_a_deadlock_failure(self):
+        exception = orm.make_deadlock_failure()
+        self.assertThat(exception, MatchesPredicate(
+            is_deadlock_failure, "%r is not a deadlock failure."))
+
+
+class TestMakeUniqueViolation(MAASTestCase):
+    """Tests for `make_unique_violation`."""
+
+    def test__makes_a_unique_violation(self):
+        exception = orm.make_unique_violation()
+        self.assertThat(exception, MatchesPredicate(
+            is_unique_violation, "%r is not a unique violation."))
+
+
 class TestRequestTransactionRetry(MAASTestCase):
     """Tests for `request_transaction_retry`."""
 
-    def test__raises_a_serialization_failure(self):
-        exception = self.assertRaises(
-            OperationalError, request_transaction_retry)
-        self.assertThat(exception, MatchesPredicate(
-            is_serialization_failure, "%r is not a serialization failure."))
+    def test__raises_a_retry_transaction_exception(self):
+        self.assertRaises(orm.RetryTransaction, request_transaction_retry)
 
 
 class TestGenRetryIntervals(MAASTestCase):