Skip to content

Don't retry a failed commit #517

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
Apr 6, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 1 addition & 9 deletions neo4j/_exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -174,14 +174,6 @@ def transaction(self):
return None


class BoltIncompleteCommitError(BoltError):
""" Raised when a disconnection occurs while still waiting for a commit
response. For non-idempotent write transactions, this leaves the data
in an unknown state with regard to whether the transaction completed
successfully or not.
"""


class BoltProtocolError(BoltError):
""" Raised when an unexpected or unsupported protocol event occurs.
"""
"""
12 changes: 11 additions & 1 deletion neo4j/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
+ RoutingServiceUnavailable
+ WriteServiceUnavailable
+ ReadServiceUnavailable
+ IncompleteCommit
+ ConfigurationError
+ AuthConfigurationError
+ CertificateConfigurationError
Expand All @@ -60,7 +61,6 @@
+ BoltConnectionBroken
+ BoltConnectionClosed
+ BoltFailure
+ BoltIncompleteCommitError
+ BoltProtocolError
+ Bolt*

Expand Down Expand Up @@ -286,6 +286,16 @@ class ReadServiceUnavailable(ServiceUnavailable):
"""


class IncompleteCommit(ServiceUnavailable):
""" Raised when the client looses connection while committing a transaction

Raised when a disconnection occurs while still waiting for a commit
response. For non-idempotent write transactions, this leaves the data
in an unknown state with regard to whether the transaction completed
successfully or not.
"""


class ResultConsumedError(DriverError):
""" Raised when trying to access records after the records have been consumed.
"""
Expand Down
17 changes: 7 additions & 10 deletions neo4j/io/_bolt3.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,18 +35,15 @@
from neo4j.meta import get_user_agent
from neo4j.exceptions import (
AuthError,
ServiceUnavailable,
DatabaseUnavailable,
NotALeader,
ConfigurationError,
ForbiddenOnReadOnlyDatabase,
IncompleteCommit,
NotALeader,
ServiceUnavailable,
SessionExpired,
ConfigurationError,
UnsupportedServerProduct,
)
from neo4j._exceptions import (
BoltIncompleteCommitError,
BoltProtocolError,
)
from neo4j._exceptions import BoltProtocolError
from neo4j.packstream import (
Unpacker,
Packer,
Expand Down Expand Up @@ -406,9 +403,9 @@ def _set_defunct(self, error=None):
for response in self.responses:
if isinstance(response, CommitResponse):
if error:
raise BoltIncompleteCommitError(message, address=None) from error
raise IncompleteCommit(message) from error
else:
raise BoltIncompleteCommitError(message, address=None)
raise IncompleteCommit(message)

if direct_driver:
if error:
Expand Down
14 changes: 6 additions & 8 deletions neo4j/io/_bolt4.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,16 +36,14 @@
from neo4j.meta import get_user_agent
from neo4j.exceptions import (
AuthError,
ServiceUnavailable,
DatabaseUnavailable,
NotALeader,
ForbiddenOnReadOnlyDatabase,
IncompleteCommit,
NotALeader,
ServiceUnavailable,
SessionExpired,
)
from neo4j._exceptions import (
BoltIncompleteCommitError,
BoltProtocolError,
)
from neo4j._exceptions import BoltProtocolError
from neo4j.packstream import (
Unpacker,
Packer,
Expand Down Expand Up @@ -417,9 +415,9 @@ def _set_defunct(self, error=None):
for response in self.responses:
if isinstance(response, CommitResponse):
if error:
raise BoltIncompleteCommitError(message, address=None) from error
raise IncompleteCommit(message) from error
else:
raise BoltIncompleteCommitError(message, address=None)
raise IncompleteCommit(message)

if direct_driver:
if error:
Expand Down
26 changes: 15 additions & 11 deletions neo4j/work/simple.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,29 +19,31 @@
# limitations under the License.


from collections import deque
from logging import getLogger
from random import random
from time import perf_counter, sleep
from warnings import warn
from time import (
perf_counter,
sleep,
)

from neo4j.conf import SessionConfig
from neo4j.api import READ_ACCESS, WRITE_ACCESS
from neo4j.data import DataHydrator, DataDehydrator
from neo4j.api import (
READ_ACCESS,
WRITE_ACCESS,
)
from neo4j.data import DataHydrator
from neo4j.exceptions import (
ClientError,
IncompleteCommit,
Neo4jError,
ServiceUnavailable,
TransientError,
SessionExpired,
TransientError,
TransactionError,
ClientError,
)
from neo4j._exceptions import BoltIncompleteCommitError
from neo4j.work import Workspace
from neo4j.work.summary import ResultSummary
from neo4j.work.transaction import Transaction
from neo4j.work.result import Result
from neo4j.io._bolt3 import Bolt3
from neo4j.work.transaction import Transaction

log = getLogger("neo4j")

Expand Down Expand Up @@ -317,6 +319,8 @@ def _run_transaction(self, access_mode, transaction_function, *args, **kwargs):
raise
else:
tx.commit()
except IncompleteCommit:
raise
except (ServiceUnavailable, SessionExpired) as error:
errors.append(error)
self._disconnect()
Expand Down
10 changes: 3 additions & 7 deletions neo4j/work/transaction.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,8 @@

from neo4j.work.result import Result
from neo4j.data import DataHydrator
from neo4j._exceptions import BoltIncompleteCommitError
from neo4j.exceptions import (
ServiceUnavailable,
IncompleteCommit,
TransactionError,
)

Expand Down Expand Up @@ -132,13 +131,10 @@ def commit(self):
self._connection.commit(on_success=metadata.update)
self._connection.send_all()
self._connection.fetch_all()
except BoltIncompleteCommitError:
self._bookmark = metadata.get("bookmark")
finally:
self._closed = True
self._on_closed()
raise ServiceUnavailable("Connection closed during commit")
self._bookmark = metadata.get("bookmark")
self._closed = True
self._on_closed()

return self._bookmark

Expand Down
2 changes: 1 addition & 1 deletion tests/unit/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
from uuid import uuid4

import neo4j.api
from neo4j.work.simple import DataDehydrator
from neo4j.data import DataDehydrator
from neo4j.exceptions import (
ConfigurationError,
)
Expand Down
2 changes: 1 addition & 1 deletion tests/unit/test_exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
RoutingServiceUnavailable,
WriteServiceUnavailable,
ReadServiceUnavailable,
IncompleteCommit,
ConfigurationError,
AuthConfigurationError,
CertificateConfigurationError,
Expand All @@ -60,7 +61,6 @@
BoltConnectionBroken,
BoltConnectionClosed,
BoltFailure,
BoltIncompleteCommitError,
BoltProtocolError,
)

Expand Down