Cascade latest release into develop

PULL_REQUEST_TEMPLATE.md

@@ -45,7 +45,7 @@ Addresses #*PR# HERE*
 - [ ] Are there any new ways this uses user data that needs to be factored into our [Privacy Policy](https://github.com/learningequality/studio/tree/master/contentcuration/contentcuration/templates/policies/text)?
 - [ ] Are there any new interactions that need to be added to the [QA Sheet](https://docs.google.com/spreadsheets/d/1HF4Gy6rb_BLbZoNkZEWZonKFBqPyVEiQq4Ve6XgIYmQ/edit#gid=0)?
 - [ ] Are there opportunities for using Google Analytics here (if applicable)?
-- [ ] If the Pipfile has been changed, is the updated Pipfile.lock file also included in this PR?
+- [ ] If any Python requirements have changed, are the updated requirements.txt files also included in this PR?
 - [ ] Are the migrations [safe for a large db](https://www.braintreepayments.com/blog/safe-operations-for-high-volume-postgresql/) (if applicable)?
 
 ## Comments

contentcuration/contentcuration/db/advisory_lock.py

@@ -0,0 +1,79 @@
+import logging as logger
+from contextlib import contextmanager
+
+from django.db import connection
+
+
+logging = logger.getLogger(__name__)
+
+
+class AdvisoryLockBusy(RuntimeError):
+    pass
+
+
+@contextmanager
+def execute_lock(key1, key2=None, unlock=False, session=False, shared=False, wait=True):
+    """
+    Creates or destroys an advisory lock within postgres
+
+    :param key1: An int sent to the PG lock function
+    :param key2: A 2nd int sent to the PG lock function
+    :param unlock: A bool representing whether query should use `unlock`
+    :param session: A bool indicating if this should persist outside of transaction
+    :param shared: A bool indicating if this should be shared, otherwise exclusive
+    :param wait: A bool indicating if it should use a `try` PG function
+    """
+    if not session:
+        if not connection.in_atomic_block:
+            raise NotImplementedError("Advisory lock requires transaction")
+        if unlock:
+            raise NotImplementedError("Transaction level locks unlock automatically")
+
+    keys = [key1]
+    if key2 is not None:
+        keys.append(key2)
+
+    query = "SELECT pg{_try}_advisory_{xact_}{lock}{_shared}({keys}) AS lock;".format(
+        _try="" if wait else "_try",
+        xact_="" if session else "xact_",
+        lock="unlock" if unlock else "lock",
+        _shared="_shared" if shared else "",
+        keys=", ".join(["%s" for i in range(0, 2 if key2 is not None else 1)])
+    )
+
+    log_query = "'{}' with params {}".format(query, keys)
+    logging.debug("Acquiring advisory lock: {}".format(query, log_query))
+    with connection.cursor() as c:
+        c.execute(query, keys)
+        logging.debug("Acquired advisory lock: {}".format(query, log_query))
+        yield c
+
+
+def advisory_lock(key1, key2=None, shared=False):
+    """
+    Creates a transaction level advisory lock that blocks until ready
+
+    :param key1: int
+    :param key2: int
+    :param shared: bool
+    """
+    with execute_lock(key1, key2=key2, shared=shared):
+        # the lock will exist until the transaction is either committed or rolled-back
+        pass
+
+
+def try_advisory_lock(key1, key2=None, shared=False):
+    """
+    Creates a transaction level advisory lock that doesn't block
+
+    :param key1: int
+    :param key2: int
+    :param shared: bool
+    :raises: AdvisoryLockBusy
+    """
+    with execute_lock(key1, key2=key2, shared=shared, wait=False) as cursor:
+        # for `try` locks, the PG function will return True or False,
+        # representing whether the lock was acquired or not
+        results = cursor.fetchone()
+        if not results[0]:
+            raise AdvisoryLockBusy("Unable to acquire advisory lock")

contentcuration/contentcuration/db/models/manager.py

@@ -12,6 +12,7 @@
 from mptt.managers import TreeManager
 from mptt.signals import node_moved
 
+from contentcuration.db.advisory_lock import advisory_lock
 from contentcuration.db.models.query import CustomTreeQuerySet
 from contentcuration.utils.tasks import increment_progress
 from contentcuration.utils.tasks import set_total
@@ -32,6 +33,7 @@
 # The exact optimum batch size is probably highly dependent on tree
 # topology also, so these rudimentary tests are likely insufficient
 BATCH_SIZE = 100
+TREE_LOCK = 1001
 
 
 class CustomManager(Manager.from_queryset(CTEQuerySet)):
@@ -81,55 +83,43 @@ def _get_next_tree_id(self, *args, **kwargs):
         return new_id
 
     @contextlib.contextmanager
-    def _attempt_lock(self, tree_ids, values):
+    def _attempt_lock(self, tree_ids, shared_tree_ids=None):
         """
         Internal method to allow the lock_mptt method to do retries in case of deadlocks
         """
+        shared_tree_ids = shared_tree_ids or []
+
         start = time.time()
         with transaction.atomic():
             # Issue a separate lock on each tree_id
             # in a predictable order.
             # This will mean that every process acquires locks in the same order
             # and should help to minimize deadlocks
             for tree_id in tree_ids:
-                execute_queryset_without_results(
-                    self.select_for_update()
-                    .order_by()
-                    .filter(tree_id=tree_id)
-                    .values(*values)
-                )
+                advisory_lock(TREE_LOCK, key2=tree_id, shared=tree_id in shared_tree_ids)
             yield
             log_lock_time_spent(time.time() - start)
 
     @contextlib.contextmanager
-    def lock_mptt(self, *tree_ids):
+    def lock_mptt(self, *tree_ids, **kwargs):
         tree_ids = sorted((t for t in set(tree_ids) if t is not None))
+        shared_tree_ids = kwargs.pop('shared_tree_ids', [])
         # If this is not inside the context of a delay context manager
         # or updates are not disabled set a lock on the tree_ids.
         if (
             not self.model._mptt_is_tracking
             and self.model._mptt_updates_enabled
             and tree_ids
         ):
-            # Lock based on MPTT columns for updates on any of the tree_ids specified
-            # until the end of this transaction
-            mptt_opts = self.model._mptt_meta
-            values = (
-                mptt_opts.tree_id_attr,
-                mptt_opts.left_attr,
-                mptt_opts.right_attr,
-                mptt_opts.level_attr,
-                mptt_opts.parent_attr,
-            )
             try:
-                with self._attempt_lock(tree_ids, values):
+                with self._attempt_lock(tree_ids, shared_tree_ids=shared_tree_ids):
                     yield
             except OperationalError as e:
                 if "deadlock detected" in e.args[0]:
                     logging.error(
                         "Deadlock detected while trying to lock ContentNode trees for mptt operations, retrying"
                     )
-                    with self._attempt_lock(tree_ids, values):
+                    with self._attempt_lock(tree_ids, shared_tree_ids=shared_tree_ids):
                         yield
                 else:
                     raise
@@ -180,7 +170,7 @@ def _mptt_refresh(self, *nodes):
         }
         for node in nodes:
             # Set the values on each of the nodes
-            if node.id:
+            if node.id and node.id in values_lookup:
                 values = values_lookup[node.id]
                 for k, v in values.items():
                     setattr(node, k, v)
@@ -556,11 +546,11 @@ def _deep_copy(
         excluded_descendants,
         can_edit_source_channel,
     ):
-
-        nodes_to_copy = self._all_nodes_to_copy(node, excluded_descendants)
+        # lock mptt source tree with shared advisory lock
+        with self.lock_mptt(node.tree_id, shared_tree_ids=[node.tree_id]):
+            nodes_to_copy = list(self._all_nodes_to_copy(node, excluded_descendants))
 
         nodes_by_parent = {}
-
         for copy_node in nodes_to_copy:
             if copy_node.parent_id not in nodes_by_parent:
                 nodes_by_parent[copy_node.parent_id] = []

contentcuration/contentcuration/forms.py

@@ -89,35 +89,35 @@ def save(self, request=None, extra_email_context=None, **kwargs):
         user.
         """
         email = self.cleaned_data["email"]
-        inactive_user = User.objects.filter(email=email, is_active=False).first()
+        user = User.get_for_email(email)
 
-        if inactive_user:
+        if user and user.is_active:
+            super(ForgotPasswordForm, self).save(request=request, extra_email_context=extra_email_context, **kwargs)
+        elif user:
             # For users who were invited but hadn't registered yet
-            if not inactive_user.password:
+            if not user.password:
                 context = {
                     'site': extra_email_context.get('site'),
-                    'user': inactive_user,
+                    'user': user,
                     'domain': extra_email_context.get('domain'),
                 }
                 subject = render_to_string('registration/password_reset_subject.txt', context)
                 subject = ''.join(subject.splitlines())
                 message = render_to_string('registration/registration_needed_email.txt', context)
-                inactive_user.email_user(subject, message, settings.DEFAULT_FROM_EMAIL, )
+                user.email_user(subject, message, settings.DEFAULT_FROM_EMAIL, )
             else:
-                activation_key = self.get_activation_key(inactive_user)
+                activation_key = self.get_activation_key(user)
                 context = {
                     'activation_key': activation_key,
                     'expiration_days': settings.ACCOUNT_ACTIVATION_DAYS,
                     'site': extra_email_context.get('site'),
-                    'user': inactive_user,
+                    'user': user,
                     'domain': extra_email_context.get('domain'),
                 }
                 subject = render_to_string('registration/password_reset_subject.txt', context)
                 subject = ''.join(subject.splitlines())
                 message = render_to_string('registration/activation_needed_email.txt', context)
-                inactive_user.email_user(subject, message, settings.DEFAULT_FROM_EMAIL, )
-        else:
-            super(ForgotPasswordForm, self).save(request=request, extra_email_context=extra_email_context, **kwargs)
+                user.email_user(subject, message, settings.DEFAULT_FROM_EMAIL, )
 
     def get_activation_key(self, user):
         """