PYTHON-1483 Prohibit unack'ed writes with explicit sessions

Author: ShaneHarvey

doc/changelog.rst

@@ -88,6 +88,13 @@ Unavoidable breaking changes:
  :class:`~pymongo.errors.OperationFailure`.
 - :meth:`~pymongo.collection.Collection.parallel_scan` no longer uses an
  implicit session. Explicit sessions are still supported.
+- Unacknowledged writes (``w=0``) with an explicit ``session`` parameter now
+ raise a client side error. Since PyMongo does not wait for a response for an
+ unacknowledged write, two unacknowledged writes run serially by the client
+ may be executed simultaneously on the server. However, the server requires a
+ single session must not be used simultaneously by more than one operation.
+ Therefore explicit sessions cannot support unacknowledged writes.
+ Unacknowledged writes without a ``session`` parameter are still supported.
 
 
 Issues Resolved

pymongo/bulk.py

@@ -23,6 +23,7 @@
 from bson.objectid import ObjectId
 from bson.raw_bson import RawBSONDocument
 from bson.son import SON
+from pymongo.client_session import _validate_session_write_concern
 from pymongo.common import (validate_is_mapping,
  validate_is_document_type,
  validate_ok_for_replace,
@@ -498,6 +499,7 @@ def execute(self, write_concern, session):
  self.executed = True
  write_concern = (WriteConcern(**write_concern) if
  write_concern else self.collection.write_concern)
+ session = _validate_session_write_concern(session, write_concern)
 
  if self.ordered:
  generator = self.gen_ordered()

pymongo/client_session.py

@@ -192,6 +192,28 @@ def read_preference(self):
  return self._read_preference
 
 
+def _validate_session_write_concern(session, write_concern):
+ """Validate that an explicit session is not used with an unack'ed write.
+
+ Returns the session to use for the next operation.
+ """
+ if session:
+ if write_concern is not None and not write_concern.acknowledged:
+ # For unacknowledged writes without an explicit session,
+ # drivers SHOULD NOT use an implicit session. If a driver
+ # creates an implicit session for unacknowledged writes
+ # without an explicit session, the driver MUST NOT send the
+ # session ID.
+ if session._implicit:
+ return None
+ else:
+ raise ConfigurationError(
+ 'Explicit sessions are incompatible with '
+ 'unacknowledged write concern: %r' % (
+ write_concern,))
+ return session
+
+
 class _TransactionContext(object):
  """Internal transaction context manager for start_transaction."""
  def __init__(self, session):
@@ -243,14 +265,16 @@ def _reraise_with_unknown_commit(exc):
 
 class ClientSession(object):
  """A session for ordering sequential operations."""
- def __init__(self, client, server_session, options, authset):
+ def __init__(self, client, server_session, options, authset, implicit):
  # A MongoClient, a _ServerSession, a SessionOptions, and a set.
  self._client = client
  self._server_session = server_session
  self._options = options
  self._authset = authset
  self._cluster_time = None
  self._operation_time = None
+ # Is this an implicitly created session?
+ self._implicit = implicit
  self._transaction = _Transaction(None)
 
  def end_session(self):

pymongo/collection.py

@@ -2258,10 +2258,6 @@ def _aggregate(self, pipeline, cursor_class, first_batch_size, session,
  if first_batch_size is not None and not dollar_out:
  kwargs["cursor"]["batchSize"] = first_batch_size
 
- if (sock_info.max_wire_version >= 5 and dollar_out and
- not self.write_concern.is_server_default):
- cmd['writeConcern'] = self.write_concern.document
-
  cmd.update(kwargs)
  # Apply this Collection's read concern if $out is not in the
  # pipeline.
@@ -2271,6 +2267,10 @@ def _aggregate(self, pipeline, cursor_class, first_batch_size, session,
  read_concern = self.read_concern
  else:
  read_concern = None
+ if 'writeConcern' not in cmd and dollar_out:
+ write_concern = self.write_concern
+ else:
+ write_concern = None
 
  # Avoid auto-injecting a session: aggregate() passes a session,
  # aggregate_raw_batches() passes none.
@@ -2282,17 +2282,18 @@ def _aggregate(self, pipeline, cursor_class, first_batch_size, session,
  self.codec_options,
  parse_write_concern_error=True,
  read_concern=read_concern,
+ write_concern=write_concern,
  collation=collation,
  session=session,
  client=self.__database.client)
 
  if "cursor" in result:
  cursor = result["cursor"]
  else:
- # Pre-MongoDB 2.6. Fake a cursor.
+ # Pre-MongoDB 2.6 or unacknowledged write. Fake a cursor.
  cursor = {
  "id": 0,
- "firstBatch": result["result"],
+ "firstBatch": result.get("result", []),
  "ns": self.full_name,
  }
 
@@ -2586,14 +2587,15 @@ def rename(self, new_name, session=None, **kwargs):
 
  new_name = "%s.%s" % (self.__database.name, new_name)
  cmd = SON([("renameCollection", self.__full_name), ("to", new_name)])
+ cmd.update(kwargs)
+ write_concern = self._write_concern_for_cmd(cmd)
+
  with self._socket_for_writes() as sock_info:
  with self.__database.client._tmp_session(session) as s:
- if (sock_info.max_wire_version >= 5 and
- not self.write_concern.is_server_default):
- cmd['writeConcern'] = self.write_concern.document
- cmd.update(kwargs)
  return sock_info.command(
- 'admin', cmd, parse_write_concern_error=True,
+ 'admin', cmd,
+ write_concern=write_concern,
+ parse_write_concern_error=True,
  session=s, client=self.__database.client)
 
  def distinct(self, key, filter=None, session=None, **kwargs):
@@ -2716,20 +2718,20 @@ def map_reduce(self, map, reduce, out, full_response=False, session=None,
  inline = 'inline' in cmd['out']
  sock_ctx, read_pref = self._socket_for_primary_reads(session)
  with sock_ctx as (sock_info, slave_ok):
- if (sock_info.max_wire_version >= 5 and
- not self.write_concern.is_server_default and
- not inline):
- cmd['writeConcern'] = self.write_concern.document
- cmd.update(kwargs)
  if (sock_info.max_wire_version >= 4 and 'readConcern' not in cmd and
  inline):
  read_concern = self.read_concern
  else:
  read_concern = None
+ if 'writeConcern' not in cmd and not inline:
+ write_concern = self.write_concern
+ else:
+ write_concern = None
 
  response = self._command(
  sock_info, cmd, slave_ok, read_pref,
  read_concern=read_concern,
+ write_concern=write_concern,
  collation=collation, session=session)
 
  if full_response or not response.get('result'):
@@ -2796,6 +2798,13 @@ def inline_map_reduce(self, map, reduce, full_response=False, session=None,
  else:
  return res.get("results")
 
+ def _write_concern_for_cmd(self, cmd):
+ raw_wc = cmd.get('writeConcern')
+ if raw_wc is not None:
+ return WriteConcern(**raw_wc)
+ else:
+ return self.write_concern
+
  def __find_and_modify(self, filter, projection, sort, upsert=None,
  return_document=ReturnDocument.BEFORE,
  array_filters=None, session=None, **kwargs):
@@ -2818,11 +2827,7 @@ def __find_and_modify(self, filter, projection, sort, upsert=None,
  common.validate_boolean("upsert", upsert)
  cmd["upsert"] = upsert
 
- write_concern = cmd.get('writeConcern')
- if write_concern is not None:
- acknowledged = write_concern.get("w") != 0
- else:
- acknowledged = self.write_concern.acknowledged
+ write_concern = self._write_concern_for_cmd(cmd)
 
  def _find_and_modify(session, sock_info, retryable_write):
  if array_filters is not None:
@@ -2835,20 +2840,20 @@ def _find_and_modify(session, sock_info, retryable_write):
  'arrayFilters is unsupported for unacknowledged '
  'writes.')
  cmd["arrayFilters"] = array_filters
- if sock_info.max_wire_version >= 4 and 'writeConcern' not in cmd:
- wc_doc = self.write_concern.document
- if wc_doc:
- cmd['writeConcern'] = wc_doc
+ if (sock_info.max_wire_version >= 4 and
+ not write_concern.is_server_default):
+ cmd['writeConcern'] = write_concern.document
  out = self._command(sock_info, cmd,
  read_preference=ReadPreference.PRIMARY,
+ write_concern=write_concern,
  allowable_errors=[_NO_OBJ_ERROR],
  collation=collation, session=session,
  retryable_write=retryable_write)
  _check_write_command_response(out)
  return out.get("value")
 
  return self.__database.client._retryable_write(
- acknowledged, _find_and_modify, session)
+ write_concern.acknowledged, _find_and_modify, session)
 
  def find_one_and_delete(self, filter,
  projection=None, sort=None, session=None, **kwargs):
@@ -3245,17 +3250,12 @@ def find_and_modify(self, query={}, update=None,
  cmd = SON([("findAndModify", self.__name)])
  cmd.update(kwargs)
 
- write_concern = cmd.get('writeConcern')
- if write_concern is not None:
- acknowledged = write_concern.get("w") != 0
- else:
- acknowledged = self.write_concern.acknowledged
+ write_concern = self._write_concern_for_cmd(cmd)
 
  def _find_and_modify(session, sock_info, retryable_write):
- if sock_info.max_wire_version >= 4 and 'writeConcern' not in cmd:
- wc_doc = self.write_concern.document
- if wc_doc:
- cmd['writeConcern'] = wc_doc
+ if (sock_info.max_wire_version >= 4 and
+ not write_concern.is_server_default):
+ cmd['writeConcern'] = write_concern.document
  result = self._command(
  sock_info, cmd, read_preference=ReadPreference.PRIMARY,
  allowable_errors=[_NO_OBJ_ERROR], collation=collation,
@@ -3265,7 +3265,7 @@ def _find_and_modify(session, sock_info, retryable_write):
  return result
 
  out = self.__database.client._retryable_write(
- acknowledged, _find_and_modify, None)
+ write_concern.acknowledged, _find_and_modify, None)
 
  if not out['ok']:
  if out["errmsg"] == _NO_OBJ_ERROR:

pymongo/database.py

@@ -498,11 +498,6 @@ def _command(self, sock_info, command, slave_ok=False, value=1, check=True,
  if isinstance(command, string_type):
  command = SON([(command, value)])
 
- if (sock_info.max_wire_version >= 5 and
- write_concern and
- not write_concern.is_server_default):
- command['writeConcern'] = write_concern.document
-
  command.update(kwargs)
  with self.__client._tmp_session(session) as s:
  return sock_info.command(
@@ -513,6 +508,7 @@ def _command(self, sock_info, command, slave_ok=False, value=1, check=True,
  codec_options,
  check,
  allowable_errors,
+ write_concern=write_concern,
  parse_write_concern_error=parse_write_concern_error,
  session=s,
  client=self.__client)

pymongo/mongo_client.py

@@ -1498,6 +1498,21 @@ def _process_periodic_tasks(self):
  except Exception:
  helpers._handle_exception()
 
+ def __start_session(self, implicit, **kwargs):
+ # Driver Sessions Spec: "If startSession is called when multiple users
+ # are authenticated drivers MUST raise an error with the error message
+ # 'Cannot call startSession when multiple users are authenticated.'"
+ authset = set(self.__all_credentials.values())
+ if len(authset) > 1:
+ raise InvalidOperation("Cannot call start_session when"
+ " multiple users are authenticated")
+
+ # Raises ConfigurationError if sessions are not supported.
+ server_session = self._get_server_session()
+ opts = client_session.SessionOptions(**kwargs)
+ return client_session.ClientSession(
+ self, server_session, opts, authset, implicit)
+
  def start_session(self,
  causal_consistency=True,
  default_transaction_options=None):
@@ -1519,21 +1534,10 @@ def start_session(self,
 
  .. versionadded:: 3.6
  """
- # Driver Sessions Spec: "If startSession is called when multiple users
- # are authenticated drivers MUST raise an error with the error message
- # 'Cannot call startSession when multiple users are authenticated.'"
- authset = set(self.__all_credentials.values())
- if len(authset) > 1:
- raise InvalidOperation("Cannot call start_session when"
- " multiple users are authenticated")
-
- # Raises ConfigurationError if sessions are not supported.
- server_session = self._get_server_session()
- opts = client_session.SessionOptions(
+ return self.__start_session(
+ False,
  causal_consistency=causal_consistency,
  default_transaction_options=default_transaction_options)
- return client_session.ClientSession(
- self, server_session, opts, authset)
 
  def _get_server_session(self):
  """Internal: start or resume a _ServerSession."""
@@ -1549,9 +1553,9 @@ def _ensure_session(self, session=None):
  return session
 
  try:
- # Don't make implied sessions causally consistent. Applications
+ # Don't make implicit sessions causally consistent. Applications
  # should always opt-in.
- return self.start_session(causal_consistency=False)
+ return self.__start_session(True, causal_consistency=False)
  except (ConfigurationError, InvalidOperation):
  # Sessions not supported, or multiple users authenticated.
  return None

pymongo/pool.py

@@ -40,6 +40,7 @@ class _SSLCertificateError(ValueError):
 from bson.py3compat import imap, itervalues, _unicode, integer_types
 from bson.son import SON
 from pymongo import auth, helpers, thread_util, __version__
+from pymongo.client_session import _validate_session_write_concern
 from pymongo.common import (MAX_BSON_SIZE,
  MAX_MESSAGE_SIZE,
  MAX_WIRE_VERSION,
@@ -534,6 +535,7 @@ def command(self, dbname, spec, slave_ok=False,
  - `publish_events`: Should we publish events for this command?
  """
  self.validate_session(client, session)
+ session = _validate_session_write_concern(session, write_concern)
 
  # Ensure command name remains in first place.
  if not isinstance(spec, ORDERED_TYPES):