Merge pull request #247 from technige/1.7-replace-ack-failure-with-reset

Replaced ACK_FAILURE with RESET

Author: technige

neo4j/bolt/connection.py

@@ -36,7 +36,7 @@
 
 from neo4j.addressing import SocketAddress, Resolver
 from neo4j.bolt.cert import KNOWN_HOSTS
-from neo4j.bolt.response import InitResponse, AckFailureResponse, ResetResponse
+from neo4j.bolt.response import InitResponse, ResetResponse
 from neo4j.compat.ssl import SSL_AVAILABLE, HAS_SNI, SSLError
 from neo4j.exceptions import ClientError, ProtocolError, SecurityError, ServiceUnavailable
 from neo4j.packstream import Packer, Unpacker
@@ -53,7 +53,6 @@
 
 # Signature bytes for each message type
 INIT = b"\x01"             # 0000 0001 // INIT <user_agent> <auth>
-ACK_FAILURE = b"\x0E"      # 0000 1110 // ACK_FAILURE
 RESET = b"\x0F"            # 0000 1111 // RESET
 RUN = b"\x10"              # 0001 0000 // RUN <statement> <parameters>
 DISCARD_ALL = b"\x2F"      # 0010 1111 // DISCARD *
@@ -240,8 +239,6 @@ def append(self, signature, fields=(), response=None):
             log_debug("C: DISCARD_ALL %r", fields)
         elif signature == RESET:
             log_debug("C: RESET %r", fields)
-        elif signature == ACK_FAILURE:
-            log_debug("C: ACK_FAILURE %r", fields)
         elif signature == INIT:
             log_debug("C: INIT (%r, {...})", fields[0])
         else:
@@ -251,13 +248,6 @@ def append(self, signature, fields=(), response=None):
         self.output_buffer.chunk()
         self.responses.append(response)
 
-    def acknowledge_failure(self):
-        """ Add an ACK_FAILURE message to the outgoing queue, send
-        it and consume all remaining messages.
-        """
-        self.append(ACK_FAILURE, response=AckFailureResponse(self))
-        self.sync()
-
     def reset(self):
         """ Add a RESET message to the outgoing queue, send
         it and consume all remaining messages.

neo4j/bolt/response.py

@@ -63,12 +63,6 @@ def on_failure(self, metadata):
             raise ServiceUnavailable(message)
 
 
-class AckFailureResponse(Response):
-
-    def on_failure(self, metadata):
-        raise ProtocolError("ACK_FAILURE failed")
-
-
 class ResetResponse(Response):
 
     def on_failure(self, metadata):

neo4j/v1/api.py

@@ -440,13 +440,19 @@ def rollback_transaction(self):
         """
         if not self.has_transaction():
             raise TransactionError("No transaction to rollback")
-        self._transaction = None
+        self._destroy_transaction()
         rollback_result = self.__rollback__()
         try:
             rollback_result.consume()
         except ServiceUnavailable:
             pass
 
+    def reset(self):
+        """ Reset the session.
+        """
+        self._destroy_transaction()
+        self._connection.reset()
+
     def _run_transaction(self, access_mode, unit_of_work, *args, **kwargs):
         if not callable(unit_of_work):
             raise TypeError("Unit of work is not callable")
@@ -640,10 +646,11 @@ def close(self):
                 self.success = False
                 raise
             finally:
-                if self.success:
-                    self.session.commit_transaction()
-                else:
-                    self.session.rollback_transaction()
+                if self.session.has_transaction():
+                    if self.success:
+                        self.session.commit_transaction()
+                    else:
+                        self.session.rollback_transaction()
                 self._closed = True
                 self.on_close()
 

neo4j/v1/result.py

@@ -65,7 +65,7 @@ def on_footer(metadata):
 
         def on_failure(metadata):
             # Called on execution failure.
-            self.session._connection.acknowledge_failure()
+            self.session.reset()
             on_footer(metadata)
             raise CypherError.hydrate(**metadata)
 

test/integration/test_session.py

@@ -357,8 +357,10 @@ def test_last_run_statement_should_be_cleared_on_failure(self):
             connection_1 = session._connection
             assert connection_1._last_run_statement == "RETURN 1"
             with self.assertRaises(CypherSyntaxError):
-                tx.run("X").consume()
-            connection_2 = session._connection
+                result = tx.run("X")
+                connection_2 = session._connection
+                result.consume()
+            # connection_2 = session._connection
             assert connection_2 is connection_1
             assert connection_2._last_run_statement is None
             tx.close()

test/stub/scripts/broken_router.script

@@ -5,5 +5,5 @@ C: RUN "CALL dbms.cluster.routing.getServers" {}
    PULL_ALL
 S: FAILURE {"code": "Neo.DatabaseError.General.UnknownError", "message": "An unknown error occurred."}
    IGNORED
-C: ACK_FAILURE
+C: RESET
 S: SUCCESS {}

test/stub/scripts/database_unavailable.script

@@ -1,7 +1,6 @@
 !: AUTO INIT
 !: AUTO RESET
 !: AUTO PULL_ALL
-!: AUTO ACK_FAILURE
 !: AUTO RUN "ROLLBACK" {}
 !: AUTO RUN "BEGIN" {}
 !: AUTO RUN "COMMIT" {}

test/stub/scripts/error_in_tx.script

@@ -11,7 +11,7 @@ C: RUN "X" {}
 S: FAILURE {"code": "Neo.ClientError.Statement.SyntaxError", "message": "X"}
    IGNORED {}
 
-C: ACK_FAILURE
+C: RESET
 S: SUCCESS {}
 
 C: RUN "ROLLBACK" {}

test/stub/scripts/forbidden_on_read_only_database.script

@@ -1,7 +1,6 @@
 !: AUTO INIT
 !: AUTO RESET
 !: AUTO PULL_ALL
-!: AUTO ACK_FAILURE
 !: AUTO RUN "ROLLBACK" {}
 !: AUTO RUN "BEGIN" {}
 !: AUTO RUN "COMMIT" {}

test/stub/scripts/non_router.script

@@ -5,5 +5,5 @@ C: RUN "CALL dbms.cluster.routing.getServers" {}
    PULL_ALL
 S: FAILURE {"code": "Neo.ClientError.Procedure.ProcedureNotFound", "message": "Not a router"}
    IGNORED
-C: ACK_FAILURE
+C: RESET
 S: SUCCESS {}