policies: prevent retry downgrade from serial to non-serial consistency

Add a guard in the retry execution path that prevents any retry policy
from downgrading SERIAL/LOCAL_SERIAL to a non-serial consistency level,
which would break serial read (Paxos) guarantees.

Also add a unit test verifying DowngradingConsistencyRetryPolicy does
not downgrade serial consistency on read timeout or unavailable.

Author: mykaul

cassandra/cluster.py

@@ -5433,7 +5433,20 @@ def _retry(self, reuse_connection, consistency_level, host, delay):
         if self._metrics is not None:
             self._metrics.on_retry()
         if consistency_level is not None:
-            self.message.consistency_level = consistency_level
+            # Never downgrade from serial to non-serial consistency, as that
+            # would break serial read (Paxos) guarantees.
+            if not ConsistencyLevel.is_serial(consistency_level):
+                original_cl = self.message.consistency_level
+                if ConsistencyLevel.is_serial(original_cl):
+                    log.debug(
+                        "Retry policy attempted to downgrade serial consistency %s to %s; "
+                        "keeping original consistency level.",
+                        ConsistencyLevel.value_to_name.get(original_cl, original_cl),
+                        ConsistencyLevel.value_to_name.get(consistency_level, consistency_level))
+                else:
+                    self.message.consistency_level = consistency_level
+            else:
+                self.message.consistency_level = consistency_level
 
         # don't retry on the event loop thread
         self.session.cluster.scheduler.schedule(delay, self._retry_task, reuse_connection, host)

tests/unit/test_policies.py

@@ -1418,6 +1418,35 @@ def test_unavailable(self):
         assert retry == RetryPolicy.RETRY
         assert consistency == ConsistencyLevel.ONE
 
+    def test_serial_consistency_not_downgraded(self):
+        """
+        Test that SERIAL/LOCAL_SERIAL consistency is never downgraded
+        to a non-serial consistency level by the retry policy.
+        @jira_ticket PYTHON-1394
+        @expected_result retry policy should rethrow or retry on next host
+                        without downgrading serial consistency
+
+        @test_category policy
+        """
+        policy = DowngradingConsistencyRetryPolicy()
+
+        for cl in (ConsistencyLevel.SERIAL, ConsistencyLevel.LOCAL_SERIAL):
+            # on_read_timeout should rethrow for serial consistency
+            retry, consistency = policy.on_read_timeout(
+                query=None, consistency=cl, required_responses=3,
+                received_responses=1, data_retrieved=True, retry_num=0)
+            assert retry == RetryPolicy.RETHROW, \
+                "Expected RETHROW for serial consistency %s on read timeout" % cl
+            assert consistency is None
+
+            # on_unavailable should retry on next host without downgrading
+            retry, consistency = policy.on_unavailable(
+                query=None, consistency=cl, required_replicas=3,
+                alive_replicas=1, retry_num=0)
+            assert retry == RetryPolicy.RETRY_NEXT_HOST, \
+                "Expected RETRY_NEXT_HOST for serial consistency %s on unavailable" % cl
+            assert consistency is None
+
 
 class ExponentialRetryPolicyTest(unittest.TestCase):
     def test_calculate_backoff(self):